improve handling of same port translation

README.md

@@ -54,11 +54,4 @@ This command **must return "yes"**
   - logsPods() function already exists
   - need to implement the logic to create each pod's logger and start the monitoring routing
 - [ ] Allow the front to known on which IP the service are reachable
-  - currently doing it by using `kubectl get nodes -o wide`
+  - currently doing it by using `kubectl get nodes -o wide`
-
-
-### Adding ingress handling to support reverse proxing
-
-- Test wether ingress-nginx is running or not
-  - Do something if not found : stop running and send error log OR start installation
-- 

conf/conf.go

@@ -3,14 +3,10 @@ package conf
 import "sync"
 
 type Config struct {
-	MongoURL	string
-	Database	string
 	LokiURL     string
-	NatsURL		string
 	ExecutionID string
 	Timeout     int
 	WorkflowID  string
-	Logs 		string
 }
 
 var instance *Config

demo_nginx/nginx.conf

@@ -0,0 +1,30 @@
+# nginx.conf
+user  nginx;
+worker_processes  auto;
+error_log  /var/log/nginx/error.log warn;
+pid        /var/run/nginx.pid;
+events {
+  worker_connections  1024;
+}
+http {
+  include       /etc/nginx/mime.types;
+  default_type  application/octet-stream;
+  log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
+  '$status $body_bytes_sent "$http_referer" '
+  '"$http_user_agent" "$http_x_forwarded_for"';
+  access_log  /var/log/nginx/access.log  main;
+  sendfile        on;
+  #tcp_nopush     on;
+  
+  keepalive_timeout  65;
+  #gzip  on;
+  #include /etc/nginx/conf.d/*.conf;
+server {
+  listen 80;
+  location / {
+    root   /usr/share/nginx/html;
+    index  index.html index.htm;
+    try_files $uri $uri/ /index.html;
+  }
+}
+}

go.mod

@@ -3,7 +3,7 @@ module oc-monitord
 go 1.22.0
 
 require (
-	cloud.o-forge.io/core/oc-lib v0.0.0-20240904135449-4f0ab6a3760f
+	cloud.o-forge.io/core/oc-lib v0.0.0-20240828135227-14d6a5f11c4e
 	github.com/akamensky/argparse v1.4.0
 	github.com/goraz/onion v0.1.3
 	github.com/nats-io/nats-server/v2 v2.10.18
@@ -42,6 +42,6 @@ require (
 	golang.org/x/crypto v0.26.0 // indirect
 	golang.org/x/net v0.28.0 // indirect
 	golang.org/x/sync v0.8.0 // indirect
-	golang.org/x/sys v0.25.0 // indirect
+	golang.org/x/sys v0.24.0 // indirect
-	golang.org/x/text v0.18.0 // indirect
+	golang.org/x/text v0.17.0 // indirect
 )

go.sum

@@ -40,8 +40,6 @@ cloud.o-forge.io/core/oc-lib v0.0.0-20240826085916-d0e1474f8f34 h1:40XQgwR9HxXSn
 cloud.o-forge.io/core/oc-lib v0.0.0-20240826085916-d0e1474f8f34/go.mod h1:1hhYh5QWAbYw9cKplQ0ZD9PMgU8t6gPqiYF8sldv1HU=
 cloud.o-forge.io/core/oc-lib v0.0.0-20240828135227-14d6a5f11c4e h1:/KWO/gIcP5f7T4r00715fNz0Y/Hil6Bj3J1ycuES1Zw=
 cloud.o-forge.io/core/oc-lib v0.0.0-20240828135227-14d6a5f11c4e/go.mod h1:FIJD0taWLJ5pjQLJ6sfE2KlTkvbmk5SMcyrxdjsaVz0=
-cloud.o-forge.io/core/oc-lib v0.0.0-20240904135449-4f0ab6a3760f h1:v9mw3uNg/DJswOvHooMu8/BMedA+vIXbma+8iUwsjUI=
-cloud.o-forge.io/core/oc-lib v0.0.0-20240904135449-4f0ab6a3760f/go.mod h1:FIJD0taWLJ5pjQLJ6sfE2KlTkvbmk5SMcyrxdjsaVz0=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/akamensky/argparse v1.4.0 h1:YGzvsTqCvbEZhL8zZu2AiA5nq805NZh75JNj4ajn1xc=
 github.com/akamensky/argparse v1.4.0/go.mod h1:S5kwC7IuDcEr5VeXtGPRVZ5o/FdhcMlQz4IZQuw64xA=
@@ -208,8 +206,6 @@ golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM=
 golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
 golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
-golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -220,8 +216,6 @@ golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
 golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
 golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc=
 golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
-golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
-golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=

main.go

@@ -20,6 +20,7 @@ import (
 
 	"cloud.o-forge.io/core/oc-lib/logs"
 	"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
+	"cloud.o-forge.io/core/oc-lib/tools"
 
 	"github.com/akamensky/argparse"
 	"github.com/google/uuid"
@@ -46,8 +47,6 @@ const localConfigFile = "./conf/local_ocmonitord_conf.json"
 
 func main() {
 
-	os.Setenv("test_service","true") 	// Only for service demo, delete before merging on main
-
 	monitorLocal = false
 	// Test if monitor is launched outside (with parameters) or in a k8s environment (env variables sets)
 	if os.Getenv("KUBERNETES_SERVICE_HOST") == "" {
@@ -62,19 +61,13 @@ func main() {
 		loadConfig(true, nil)
 	}
 
-	logger = logs.CreateLogger("oc-monitord")
+	logs.SetAppName("oc-monitord")
+	logger = logs.CreateLogger("oc-monitord", conf.GetConfig().LokiURL)
 
 	logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
 	logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
 
-	oclib.SetConfig(
+	oclib.Init("oc-monitord","","")
-		conf.GetConfig().MongoURL,
-		conf.GetConfig().Database,
-		conf.GetConfig().NatsURL,
-		conf.GetConfig().LokiURL,
-		conf.GetConfig().Logs,
-	)
-	oclib.Init("oc-monitord")
 
 	wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
 	conf.GetConfig().WorkflowID = wf_id
@@ -229,9 +222,7 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
 		conf.GetConfig().ExecutionID = o.GetString("workflow")
 		mongo := o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
 		db := o.GetStringDefault("database", "DC_myDC")
-
+		tools.SetConfig(mongo, db, "")
-		conf.GetConfig().MongoURL = mongo
-		conf.GetConfig().Database = db
 	} else {
 		url := parser.String("u", "url", &argparse.Options{Required: true, Default: "http://127.0.0.1:3100", Help: "Url to the Loki database logs will be sent to"})
 		execution := parser.String("e", "execution", &argparse.Options{Required: true, Help: "Execution ID of the workflow to request from oc-catalog API"})
@@ -243,21 +234,17 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
 			fmt.Println(parser.Usage(err))
 			os.Exit(1)
 		}
-		
-		conf.GetConfig().Logs = "debug"
-
 		conf.GetConfig().LokiURL = *url
-		conf.GetConfig().MongoURL = *mongo
-		conf.GetConfig().Database = *db
 		conf.GetConfig().Timeout = *timeout
 		conf.GetConfig().ExecutionID = *execution
+		tools.SetConfig(*mongo, *db, "")
 	}
 
 }
 
 
 func initOnion(o *onion.Onion) *onion.Onion {
-	logger = logs.CreateLogger("oc-monitord")
+	logger = logs.CreateLogger("oc-monitord", "")
 	configFile := ""
 
 	l3 := onion.NewEnvLayerPrefix("_", "OCMONITORD")

models/services.go

@@ -17,7 +17,7 @@ type Service struct {
 }
 
 type Metadata struct {
-    Name      string            `yaml:"name"`
+    Name      string            `yaml:"generateName"`
     
 }
 

workflow_builder/argo_builder.go

@@ -5,9 +5,11 @@
 package workflow_builder
 
 import (
+	"oc-monitord/models"
 	. "oc-monitord/models"
 	"os"
 	"slices"
+	"strconv"
 	"strings"
 	"time"
 
@@ -25,7 +27,7 @@ var logger zerolog.Logger
 type ArgoBuilder struct {
 	OriginWorkflow 	w.Workflow
 	Workflow       	Workflow
-	Services		*Service
+	Services		[]Service
 	Timeout        	int
 }
 
@@ -49,10 +51,7 @@ type Spec struct {
 func (b *ArgoBuilder) CreateDAG() (string, error) {
 	
 	// handle services by checking if there is only one processing with hostname and port
-	
+		
-	b.createNginxVolumes()
-	
-	
 	b.createTemplates()
 	b.createDAGstep()
 	b.createVolumes()
@@ -91,6 +90,7 @@ func (b *ArgoBuilder) createTemplates() {
 		var command string
 		var args string
 		var env string
+		var serv models.Service
 
 		comp_res := comp.Processing
 
@@ -114,24 +114,22 @@ func (b *ArgoBuilder) createTemplates() {
 		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)
+			serv = b.CreateService(comp)
-			b.createService(serv, argo_name, comp.ID)
+			b.addServiceToWorkflow(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
+			new_temp.Metadata.Labels["app"] = serv.Spec.Selector["app"]		// Construct the template for the k8s service and add a link in graph between k8s service and processing
+			b.addServiceToArgo(serv)
 			// if err != nil {
 			// 	// TODO
 			// }
 		}
 		
 		b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, new_temp)
+
 	}
 
-	if b.Services != nil {
-		b.addServiceToArgo()
-	}
 
 }
 
@@ -153,8 +151,9 @@ func (b *ArgoBuilder) createDAGstep() {
 		new_dag.Tasks = append(new_dag.Tasks, step)
 	}
 
-	if b.Services != nil {
+	for i, _ := range b.Services {
-		new_dag.Tasks = append(new_dag.Tasks, Task{Name:"workflow-service-pod", Template: "workflow-service-pod"})
+		name := "workflow-service-pod-"+strconv.Itoa(i + 1)
+		new_dag.Tasks = append(new_dag.Tasks, Task{Name: name , Template: name})
 	}
 
 	b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, Template{Name: "dag", Dag: new_dag})
@@ -170,15 +169,6 @@ func (b *ArgoBuilder) createVolumes() {
 	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) {
@@ -320,13 +310,4 @@ func (b *ArgoBuilder) isService(id string) bool{
 }
 
 
-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
-		}
-	}
-}
 

workflow_builder/argo_services.go

@@ -7,6 +7,7 @@ import (
 
 	"cloud.o-forge.io/core/oc-lib/models/resource_model"
 	"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
+	"github.com/nwtgck/go-fakelish"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"gopkg.in/yaml.v3"
@@ -60,10 +61,10 @@ func (b *ArgoBuilder) CreateService(processing graph.GraphItem) models.Service{
 	new_service := models.Service{APIVersion: "v1", 
 							Kind: "Service", 
 							Metadata: models.Metadata{
-								Name: "workflow-service" ,
+								Name: "workflow-service-" ,
 							}, 
 							Spec: models.ServiceSpec{
-								Selector: map[string]string{"app": "oc-service"},
+								Selector: map[string]string{"app": "service-" + fakelish.GenerateFakeWord(5, 8)},
 								Ports: []models.ServicePort{
 								},
 								Type: "NodePort",
@@ -115,25 +116,28 @@ func completeServicePorts(service *models.Service, processing graph.GraphItem) {
 	return
 }
 
-func (b *ArgoBuilder) createService(service models.Service, processing_name string, processing_id string) {
+// The k8s service passed as the parameter only expose one port because it is the result of CreateService()
-	if b.Services != nil{
+// we check if this port is already exposed by a service in the workflow and we proceed to the creation of a new service OR 
-		b.Services.Spec.Ports = append(b.Services.Spec.Ports, service.Spec.Ports...)
+// add the port to the list of port exposed by an existing if portAlreadyExposed is false 
-		}else {
+func (b *ArgoBuilder) addServiceToWorkflow(service models.Service, processing_name string, processing_id string) (label string) {
-			b.Services = &service
+	if exposed, service_available_port := b.portAlreadyExposed(service.Spec.Ports[0].TargetPort); exposed && service_available_port != nil{
-		}
+		// The port you want to expose is already exposed by all the existing services
-		
+		service_available_port.Spec.Ports = append(service_available_port.Spec.Ports, service.Spec.Ports...)
-	b.addLabel(processing_name,processing_id)
+		return service_available_port.Spec.Selector["app"]
+	} 
 
+	b.Services = append(b.Services, service)
+	return service.Spec.Selector["app"]	
 }
 
-func (b *ArgoBuilder) addServiceToArgo() error {
+func (b *ArgoBuilder) addServiceToArgo(service models.Service) error {
-	service_manifest, err := yaml.Marshal(b.Services)
+	service_manifest, err := yaml.Marshal(service)
 	if err != nil {
 		logger.Error().Msg("Could not marshal service manifest")
 		return err
 	}
 	
-	service_template := models.Template{Name: "workflow-service-pod", 
+	service_template := models.Template{Name: "workflow-service-pod-" + strconv.Itoa(len(b.Services)), 
 								Resource: models.ServiceResource{
 									Action: "create", 
 									SuccessCondition: "status.succeeded > 0",
@@ -145,4 +149,33 @@ func (b *ArgoBuilder) addServiceToArgo() error {
 	b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, service_template)
 
 	return nil
+}
+
+
+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) portAlreadyExposed(port int64) (exposed bool, service *models.Service ){
+	// For all already existing k8s services, test if the port in parameter is already exposed and returns the first service that doesn't yet expose this port 
+	for _, s := range b.Services {
+		i := 0
+		port_exposed := false
+		for !port_exposed {
+			if s.Spec.Ports[i].TargetPort == port {
+				port_exposed = true
+			}
+		}
+		if !port_exposed {
+			return false, &s
+		}
+	}
+
+	return true, nil
 }