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

change for the win

Browse Source
This commit is contained in:
mr 2024-09-25 15:46:39 +02:00
parent 3e4ac40109
commit 42375a60cb
3 changed files with 31 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
main.go
View File

@ -62,11 +62,7 @@ func main() {
monitorLocal = true monitorLocal = true
loadConfig(true, nil) loadConfig(true, nil)
} }
oclib.Init("oc-monitord")
logger = logs.CreateLogger("oc-monitord")
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
oclib.SetConfig( oclib.SetConfig(
conf.GetConfig().MongoURL, conf.GetConfig().MongoURL,
@ -75,7 +71,11 @@ func main() {
conf.GetConfig().LokiURL, conf.GetConfig().LokiURL,
conf.GetConfig().Logs, conf.GetConfig().Logs,
) )
oclib.Init("oc-monitord")
logger = logs.CreateLogger("oc-monitord")
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
wf_id := getWorkflowId(conf.GetConfig().ExecutionID) wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
conf.GetConfig().WorkflowID = wf_id conf.GetConfig().WorkflowID = wf_id

BIN
oc-monitord
View File

Binary file not shown.

33
workflow_builder/argo_builder.go
View File

@ -5,6 +5,7 @@
package workflow_builder package workflow_builder
import ( import (
"fmt"
. "oc-monitord/models" . "oc-monitord/models"
"os" "os"
"slices" "slices"
@ -23,10 +24,10 @@ import (
var logger zerolog.Logger 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
} }
type Workflow struct { type Workflow struct {
@ -52,12 +53,10 @@ func (b *ArgoBuilder) CreateDAG() (string, error) {
b.createNginxVolumes() b.createNginxVolumes()
b.createTemplates() b.createTemplates()
b.createDAGstep() b.createDAGstep()
b.createVolumes() b.createVolumes()
if b.Timeout > 0 { if b.Timeout > 0 {
b.Workflow.Spec.Timeout = b.Timeout b.Workflow.Spec.Timeout = b.Timeout
} }
@ -113,14 +112,14 @@ func (b *ArgoBuilder) createTemplates() {
argo_name := getArgoName(comp_res.GetName(), comp.ID) argo_name := getArgoName(comp_res.GetName(), comp.ID)
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 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.createService(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"] = "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"] = "oc-service" // Construct the template for the k8s service and add a link in graph between k8s service and processing
// if err != nil { // if err != nil {
// // TODO // // TODO
// } // }
@ -154,7 +153,7 @@ func (b *ArgoBuilder) createDAGstep() {
} }
if b.Services != nil { if b.Services != nil {
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: "workflow-service-pod", Template: "workflow-service-pod"})
} }
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})
@ -180,13 +179,13 @@ func (b *ArgoBuilder) createNginxVolumes() {
b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, new_volume) 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) {
for _, link := range b.OriginWorkflow.Graph.Links { for _, link := range b.OriginWorkflow.Graph.Links {
if b.OriginWorkflow.Graph.Items[link.Source.ID].Processing == nil { if b.OriginWorkflow.Graph.Items[link.Source.ID].Processing == nil {
continue continue
} }
source := b.OriginWorkflow.Graph.Items[link.Source.ID].Processing source := b.OriginWorkflow.Graph.Items[link.Source.ID].Processing
fmt.Println("source", source, current_computing_id, link.Destination.ID)
if current_computing_id == link.Destination.ID && source != nil { if current_computing_id == link.Destination.ID && source != nil {
dependency_name := getArgoName(source.GetName(), link.Source.ID) dependency_name := getArgoName(source.GetName(), link.Source.ID)
dependencies = append(dependencies, dependency_name) dependencies = append(dependencies, dependency_name)
@ -209,7 +208,7 @@ func getComputingArgs(user_input string, command string) (list_args []string) {
return return
} }
args := strings.Split(user_input," ") args := strings.Split(user_input, " ")
// quickfix that might need improvement // quickfix that might need improvement
if strings.Contains(command, "sh -c") { if strings.Contains(command, "sh -c") {
@ -307,7 +306,7 @@ func getStringValue(comp resource_model.AbstractResource, key string) string {
return "" return ""
} }
func (b *ArgoBuilder) isService(id string) bool{ func (b *ArgoBuilder) isService(id string) bool {
comp := b.OriginWorkflow.Graph.Items[id] comp := b.OriginWorkflow.Graph.Items[id]
@ -319,14 +318,12 @@ func (b *ArgoBuilder) isService(id string) bool{
return is_exposed return is_exposed
} }
func (b *ArgoBuilder) addLabel(name string, id string) { func (b *ArgoBuilder) addLabel(name string, id string) {
argo_name := getArgoName(name,id) argo_name := getArgoName(name, id)
for _, template := range b.Workflow.Spec.Templates{ for _, template := range b.Workflow.Spec.Templates {
if template.Name == argo_name{ if template.Name == argo_name {
template.Metadata.Labels["app"] = "service-workflow" template.Metadata.Labels["app"] = "service-workflow"
return return
} }
} }
} }