Compare commits
3 Commits
services_d
...
main
Author | SHA1 | Date | |
---|---|---|---|
3a7f3f6685 | |||
3d39455545 | |||
072be43b85 |
@ -55,3 +55,10 @@ This command **must return "yes"**
|
|||||||
- need to implement the logic to create each pod's logger and start the monitoring routing
|
- 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
|
- [ ] 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
|
||||||
|
-
|
@ -3,10 +3,14 @@ package conf
|
|||||||
import "sync"
|
import "sync"
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
|
MongoURL string
|
||||||
|
Database string
|
||||||
LokiURL string
|
LokiURL string
|
||||||
|
NatsURL string
|
||||||
ExecutionID string
|
ExecutionID string
|
||||||
Timeout int
|
Timeout int
|
||||||
WorkflowID string
|
WorkflowID string
|
||||||
|
Logs string
|
||||||
}
|
}
|
||||||
|
|
||||||
var instance *Config
|
var instance *Config
|
||||||
|
@ -1,30 +0,0 @@
|
|||||||
# 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;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
6
go.mod
6
go.mod
@ -3,7 +3,7 @@ module oc-monitord
|
|||||||
go 1.22.0
|
go 1.22.0
|
||||||
|
|
||||||
require (
|
require (
|
||||||
cloud.o-forge.io/core/oc-lib v0.0.0-20240828135227-14d6a5f11c4e
|
cloud.o-forge.io/core/oc-lib v0.0.0-20240904135449-4f0ab6a3760f
|
||||||
github.com/akamensky/argparse v1.4.0
|
github.com/akamensky/argparse v1.4.0
|
||||||
github.com/goraz/onion v0.1.3
|
github.com/goraz/onion v0.1.3
|
||||||
github.com/nats-io/nats-server/v2 v2.10.18
|
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/crypto v0.26.0 // indirect
|
||||||
golang.org/x/net v0.28.0 // indirect
|
golang.org/x/net v0.28.0 // indirect
|
||||||
golang.org/x/sync v0.8.0 // indirect
|
golang.org/x/sync v0.8.0 // indirect
|
||||||
golang.org/x/sys v0.24.0 // indirect
|
golang.org/x/sys v0.25.0 // indirect
|
||||||
golang.org/x/text v0.17.0 // indirect
|
golang.org/x/text v0.18.0 // indirect
|
||||||
)
|
)
|
||||||
|
6
go.sum
6
go.sum
@ -40,6 +40,8 @@ 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-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 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-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/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 h1:YGzvsTqCvbEZhL8zZu2AiA5nq805NZh75JNj4ajn1xc=
|
||||||
github.com/akamensky/argparse v1.4.0/go.mod h1:S5kwC7IuDcEr5VeXtGPRVZ5o/FdhcMlQz4IZQuw64xA=
|
github.com/akamensky/argparse v1.4.0/go.mod h1:S5kwC7IuDcEr5VeXtGPRVZ5o/FdhcMlQz4IZQuw64xA=
|
||||||
@ -206,6 +208,8 @@ 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.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 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
|
||||||
golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
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-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
|
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=
|
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||||
@ -216,6 +220,8 @@ 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.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 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc=
|
||||||
golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
|
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-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-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
|
||||||
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||||
|
27
main.go
27
main.go
@ -20,7 +20,6 @@ import (
|
|||||||
|
|
||||||
"cloud.o-forge.io/core/oc-lib/logs"
|
"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/models/workflow_execution"
|
||||||
"cloud.o-forge.io/core/oc-lib/tools"
|
|
||||||
|
|
||||||
"github.com/akamensky/argparse"
|
"github.com/akamensky/argparse"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
@ -47,6 +46,8 @@ const localConfigFile = "./conf/local_ocmonitord_conf.json"
|
|||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|
||||||
|
os.Setenv("test_service","true") // Only for service demo, delete before merging on main
|
||||||
|
|
||||||
monitorLocal = false
|
monitorLocal = false
|
||||||
// Test if monitor is launched outside (with parameters) or in a k8s environment (env variables sets)
|
// Test if monitor is launched outside (with parameters) or in a k8s environment (env variables sets)
|
||||||
if os.Getenv("KUBERNETES_SERVICE_HOST") == "" {
|
if os.Getenv("KUBERNETES_SERVICE_HOST") == "" {
|
||||||
@ -61,13 +62,19 @@ func main() {
|
|||||||
loadConfig(true, nil)
|
loadConfig(true, nil)
|
||||||
}
|
}
|
||||||
|
|
||||||
logs.SetAppName("oc-monitord")
|
logger = logs.CreateLogger("oc-monitord")
|
||||||
logger = logs.CreateLogger("oc-monitord", conf.GetConfig().LokiURL)
|
|
||||||
|
|
||||||
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
|
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
|
||||||
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
|
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
|
||||||
|
|
||||||
oclib.Init("oc-monitord","","")
|
oclib.SetConfig(
|
||||||
|
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)
|
wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
|
||||||
conf.GetConfig().WorkflowID = wf_id
|
conf.GetConfig().WorkflowID = wf_id
|
||||||
@ -222,7 +229,9 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
|
|||||||
conf.GetConfig().ExecutionID = o.GetString("workflow")
|
conf.GetConfig().ExecutionID = o.GetString("workflow")
|
||||||
mongo := o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
|
mongo := o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
|
||||||
db := o.GetStringDefault("database", "DC_myDC")
|
db := o.GetStringDefault("database", "DC_myDC")
|
||||||
tools.SetConfig(mongo, db, "")
|
|
||||||
|
conf.GetConfig().MongoURL = mongo
|
||||||
|
conf.GetConfig().Database = db
|
||||||
} else {
|
} 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"})
|
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"})
|
execution := parser.String("e", "execution", &argparse.Options{Required: true, Help: "Execution ID of the workflow to request from oc-catalog API"})
|
||||||
@ -234,17 +243,21 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
|
|||||||
fmt.Println(parser.Usage(err))
|
fmt.Println(parser.Usage(err))
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
conf.GetConfig().Logs = "debug"
|
||||||
|
|
||||||
conf.GetConfig().LokiURL = *url
|
conf.GetConfig().LokiURL = *url
|
||||||
|
conf.GetConfig().MongoURL = *mongo
|
||||||
|
conf.GetConfig().Database = *db
|
||||||
conf.GetConfig().Timeout = *timeout
|
conf.GetConfig().Timeout = *timeout
|
||||||
conf.GetConfig().ExecutionID = *execution
|
conf.GetConfig().ExecutionID = *execution
|
||||||
tools.SetConfig(*mongo, *db, "")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
func initOnion(o *onion.Onion) *onion.Onion {
|
func initOnion(o *onion.Onion) *onion.Onion {
|
||||||
logger = logs.CreateLogger("oc-monitord", "")
|
logger = logs.CreateLogger("oc-monitord")
|
||||||
configFile := ""
|
configFile := ""
|
||||||
|
|
||||||
l3 := onion.NewEnvLayerPrefix("_", "OCMONITORD")
|
l3 := onion.NewEnvLayerPrefix("_", "OCMONITORD")
|
||||||
|
@ -17,7 +17,7 @@ type Service struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type Metadata struct {
|
type Metadata struct {
|
||||||
Name string `yaml:"generateName"`
|
Name string `yaml:"name"`
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -5,11 +5,9 @@
|
|||||||
package workflow_builder
|
package workflow_builder
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"oc-monitord/models"
|
|
||||||
. "oc-monitord/models"
|
. "oc-monitord/models"
|
||||||
"os"
|
"os"
|
||||||
"slices"
|
"slices"
|
||||||
"strconv"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@ -27,7 +25,7 @@ var logger zerolog.Logger
|
|||||||
type ArgoBuilder struct {
|
type ArgoBuilder struct {
|
||||||
OriginWorkflow w.Workflow
|
OriginWorkflow w.Workflow
|
||||||
Workflow Workflow
|
Workflow Workflow
|
||||||
Services []Service
|
Services *Service
|
||||||
Timeout int
|
Timeout int
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -52,6 +50,9 @@ func (b *ArgoBuilder) CreateDAG() (string, error) {
|
|||||||
|
|
||||||
// handle services by checking if there is only one processing with hostname and port
|
// handle services by checking if there is only one processing with hostname and port
|
||||||
|
|
||||||
|
b.createNginxVolumes()
|
||||||
|
|
||||||
|
|
||||||
b.createTemplates()
|
b.createTemplates()
|
||||||
b.createDAGstep()
|
b.createDAGstep()
|
||||||
b.createVolumes()
|
b.createVolumes()
|
||||||
@ -90,7 +91,6 @@ func (b *ArgoBuilder) createTemplates() {
|
|||||||
var command string
|
var command string
|
||||||
var args string
|
var args string
|
||||||
var env string
|
var env string
|
||||||
var serv models.Service
|
|
||||||
|
|
||||||
comp_res := comp.Processing
|
comp_res := comp.Processing
|
||||||
|
|
||||||
@ -114,22 +114,24 @@ func (b *ArgoBuilder) createTemplates() {
|
|||||||
new_temp := Template{Name: argo_name, Container: temp_container}
|
new_temp := Template{Name: argo_name, Container: temp_container}
|
||||||
new_temp.Inputs.Parameters = inputs_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: "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)){
|
if (b.isService(comp.ID)){
|
||||||
serv = b.CreateService(comp)
|
serv := b.CreateService(comp)
|
||||||
b.addServiceToWorkflow(serv, argo_name, comp.ID)
|
b.createService(serv, argo_name, comp.ID)
|
||||||
new_temp.Metadata.Labels = make(map[string]string)
|
new_temp.Metadata.Labels = make(map[string]string)
|
||||||
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
|
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
|
||||||
b.addServiceToArgo(serv)
|
|
||||||
// if err != nil {
|
// if err != nil {
|
||||||
// // TODO
|
// // TODO
|
||||||
// }
|
// }
|
||||||
}
|
}
|
||||||
|
|
||||||
b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, new_temp)
|
b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, new_temp)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if b.Services != nil {
|
||||||
|
b.addServiceToArgo()
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -151,9 +153,8 @@ func (b *ArgoBuilder) createDAGstep() {
|
|||||||
new_dag.Tasks = append(new_dag.Tasks, step)
|
new_dag.Tasks = append(new_dag.Tasks, step)
|
||||||
}
|
}
|
||||||
|
|
||||||
for i, _ := range b.Services {
|
if b.Services != nil {
|
||||||
name := "workflow-service-pod-"+strconv.Itoa(i + 1)
|
new_dag.Tasks = append(new_dag.Tasks, Task{Name:"workflow-service-pod", Template: "workflow-service-pod"})
|
||||||
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})
|
b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, Template{Name: "dag", Dag: new_dag})
|
||||||
@ -169,6 +170,15 @@ func (b *ArgoBuilder) createVolumes() {
|
|||||||
b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, new_volume)
|
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) {
|
func (b *ArgoBuilder) getDependency(current_computing_id string) (dependencies []string) {
|
||||||
@ -310,4 +320,13 @@ 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
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@ -7,7 +7,6 @@ import (
|
|||||||
|
|
||||||
"cloud.o-forge.io/core/oc-lib/models/resource_model"
|
"cloud.o-forge.io/core/oc-lib/models/resource_model"
|
||||||
"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
|
"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"
|
||||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||||
"gopkg.in/yaml.v3"
|
"gopkg.in/yaml.v3"
|
||||||
@ -61,10 +60,10 @@ func (b *ArgoBuilder) CreateService(processing graph.GraphItem) models.Service{
|
|||||||
new_service := models.Service{APIVersion: "v1",
|
new_service := models.Service{APIVersion: "v1",
|
||||||
Kind: "Service",
|
Kind: "Service",
|
||||||
Metadata: models.Metadata{
|
Metadata: models.Metadata{
|
||||||
Name: "workflow-service-" ,
|
Name: "workflow-service" ,
|
||||||
},
|
},
|
||||||
Spec: models.ServiceSpec{
|
Spec: models.ServiceSpec{
|
||||||
Selector: map[string]string{"app": "service-" + fakelish.GenerateFakeWord(5, 8)},
|
Selector: map[string]string{"app": "oc-service"},
|
||||||
Ports: []models.ServicePort{
|
Ports: []models.ServicePort{
|
||||||
},
|
},
|
||||||
Type: "NodePort",
|
Type: "NodePort",
|
||||||
@ -116,28 +115,25 @@ func completeServicePorts(service *models.Service, processing graph.GraphItem) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// The k8s service passed as the parameter only expose one port because it is the result of CreateService()
|
func (b *ArgoBuilder) createService(service models.Service, processing_name string, processing_id string) {
|
||||||
// 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
|
if b.Services != nil{
|
||||||
// add the port to the list of port exposed by an existing if portAlreadyExposed is false
|
b.Services.Spec.Ports = append(b.Services.Spec.Ports, service.Spec.Ports...)
|
||||||
func (b *ArgoBuilder) addServiceToWorkflow(service models.Service, processing_name string, processing_id string) (label string) {
|
}else {
|
||||||
if exposed, service_available_port := b.portAlreadyExposed(service.Spec.Ports[0].TargetPort); exposed && service_available_port != nil{
|
b.Services = &service
|
||||||
// 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...)
|
|
||||||
return service_available_port.Spec.Selector["app"]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
b.Services = append(b.Services, service)
|
b.addLabel(processing_name,processing_id)
|
||||||
return service.Spec.Selector["app"]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *ArgoBuilder) addServiceToArgo(service models.Service) error {
|
func (b *ArgoBuilder) addServiceToArgo() error {
|
||||||
service_manifest, err := yaml.Marshal(service)
|
service_manifest, err := yaml.Marshal(b.Services)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Error().Msg("Could not marshal service manifest")
|
logger.Error().Msg("Could not marshal service manifest")
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
service_template := models.Template{Name: "workflow-service-pod-" + strconv.Itoa(len(b.Services)),
|
service_template := models.Template{Name: "workflow-service-pod",
|
||||||
Resource: models.ServiceResource{
|
Resource: models.ServiceResource{
|
||||||
Action: "create",
|
Action: "create",
|
||||||
SuccessCondition: "status.succeeded > 0",
|
SuccessCondition: "status.succeeded > 0",
|
||||||
@ -150,32 +146,3 @@ func (b *ArgoBuilder) addServiceToArgo(service models.Service) error {
|
|||||||
|
|
||||||
return nil
|
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
|
|
||||||
}
|
|
Loading…
Reference in New Issue
Block a user