diff --git a/main.go b/main.go index 2fac24e..6b5300d 100644 --- a/main.go +++ b/main.go @@ -87,7 +87,7 @@ func main() { logger.Error().Msg("Could not retrieve workflow " + conf.GetConfig().WorkflowID + " from oc-catalog API") } - argo_file_path, stepMax, err := new_wf.ExportToArgo(exec.ExecutionsID, conf.GetConfig().Timeout) + builder, argo_file_path, stepMax, err := new_wf.ExportToArgo(exec.ExecutionsID, conf.GetConfig().Timeout) if err != nil { logger.Error().Msg("Could not create the Argo file for " + conf.GetConfig().WorkflowID) logger.Error().Msg(err.Error()) @@ -99,15 +99,20 @@ func main() { wf_logger = logger.With().Str("argo_name", workflowName).Str("workflow_id", conf.GetConfig().WorkflowID).Str("workflow_execution_id", conf.GetConfig().ExecutionID).Logger() wf_logger.Debug().Msg("Testing argo name") - if conf.GetConfig().KubeHost == "" { - // Not in a k8s environment, get conf from parameters - fmt.Println("Executes outside of k8s") - executeOutside(argo_file_path, stepMax) - } else { - // Executed in a k8s environment - fmt.Println("Executes inside a k8s") - executeInside(exec.GetID(), "argo", argo_file_path, stepMax) + err = builder.CompleteBuild(exec.ExecutionsID) + if err != nil { + logger.Error().Msg(err.Error()) } + _ = stepMax + // if conf.GetConfig().KubeHost == "" { + // // Not in a k8s environment, get conf from parameters + // fmt.Println("Executes outside of k8s") + // executeOutside(argo_file_path, stepMax) + // } else { + // // Executed in a k8s environment + // fmt.Println("Executes inside a k8s") + // executeInside(exec.GetID(), "argo", argo_file_path, stepMax) + // } } // So far we only log the output from @@ -224,9 +229,9 @@ func loadConfig(is_k8s bool, parser *argparse.Parser) { o = initOnion(o) setConf(is_k8s, o, parser) - if !IsValidUUID(conf.GetConfig().ExecutionID) { - logger.Fatal().Msg("Provided ID is not an UUID") - } + // if !IsValidUUID(conf.GetConfig().ExecutionID) { + // logger.Fatal().Msg("Provided ID is not an UUID") + // } } func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) { diff --git a/models/template.go b/models/template.go index 4d22cb6..8398a6c 100644 --- a/models/template.go +++ b/models/template.go @@ -95,7 +95,7 @@ type Template struct { Resource ServiceResource `yaml:"resource,omitempty"` } -func (template *Template) CreateContainer(processing *resources.ProcessingResource, dag *Dag, isReparted bool, reparted_url string) { +func (template *Template) CreateContainer(processing *resources.ProcessingResource, dag *Dag) { instance := processing.GetSelectedInstance() if instance == nil { return @@ -116,7 +116,7 @@ func (template *Template) CreateContainer(processing *resources.ProcessingResour template.Outputs.Parameters = append(template.Inputs.Parameters, Parameter{Name: v.Name}) } cmd := strings.ReplaceAll(inst.Access.Container.Command, container.Image, "") - container.Args = append(container.Args, "echo "+templateName+" && ") // a casual echo to know where we are for logs purpose + for _, a := range strings.Split(cmd, " ") { container.Args = append(container.Args, template.ReplacePerEnv(a, inst.Env)) } @@ -125,10 +125,6 @@ func (template *Template) CreateContainer(processing *resources.ProcessingResour } container.Args = []string{strings.Join(container.Args, " ")} - if isReparted { - - } - template.Container = container } diff --git a/workflow_builder/admiralty_setter.go b/workflow_builder/admiralty_setter.go index 23ba40a..6e222bb 100644 --- a/workflow_builder/admiralty_setter.go +++ b/workflow_builder/admiralty_setter.go @@ -1,2 +1,53 @@ package workflow_builder +import ( + "fmt" + + oclib "cloud.o-forge.io/core/oc-lib" + "cloud.o-forge.io/core/oc-lib/tools" +) + +type AdmiraltySetter struct { + Id string +} + +func (s *AdmiraltySetter) InitializeAdmiralty(localPeerID string,remotePeerID string) error { + + data := oclib.NewRequest(oclib.LibDataEnum(oclib.PEER),"",localPeerID,nil,nil).LoadOne(remotePeerID) + if data.Code != 200 { + logger.Error().Msg("Error while trying to instantiate remote peer " + remotePeerID) + return fmt.Errorf(data.Err) + } + remotePeer := data.ToPeer() + + caller := tools.NewHTTPCaller( + map[tools.DataType]map[tools.METHOD]string{ + tools.ADMIRALTY_SOURCE: map[tools.METHOD]string{ + tools.POST : "/:id", + }, + tools.ADMIRALTY_KUBECONFIG: map[tools.METHOD]string{ + tools.POST: "/:id", + }, + tools.ADMIRALTY_SECRET: map[tools.METHOD]string{ + tools.POST: "/:id", + }, + tools.ADMIRALTY_TARGET: map[tools.METHOD]string{ + tools.POST: "/:id", + }, + tools.ADMIRALTY_NODES: map[tools.METHOD]string{ + tools.GET: "/id", + }, + }, + ) + fmt.Println("Creating source in ") + resp, err := remotePeer.LaunchPeerExecution(remotePeer.UUID,"toto-5",tools.ADMIRALTY_SOURCE,tools.POST,nil,caller) + if err != nil { + fmt.Println("Error contacting remote peer") + fmt.Println(err) + panic(0) + } + + fmt.Println(resp) + + return nil +} \ No newline at end of file diff --git a/workflow_builder/argo_builder.go b/workflow_builder/argo_builder.go index 4990ecd..dd90272 100644 --- a/workflow_builder/argo_builder.go +++ b/workflow_builder/argo_builder.go @@ -17,7 +17,6 @@ import ( "cloud.o-forge.io/core/oc-lib/models/common/enum" "cloud.o-forge.io/core/oc-lib/models/resources" w "cloud.o-forge.io/core/oc-lib/models/workflow" - "cloud.o-forge.io/core/oc-lib/tools" "github.com/nwtgck/go-fakelish" "github.com/rs/zerolog" "gopkg.in/yaml.v3" @@ -26,10 +25,11 @@ import ( var logger zerolog.Logger type ArgoBuilder struct { - OriginWorkflow *w.Workflow - Workflow Workflow - Services []*Service - Timeout int + OriginWorkflow *w.Workflow + Workflow Workflow + Services []*Service + Timeout int + RemotePeers []string } type Workflow struct { @@ -188,8 +188,11 @@ func (b *ArgoBuilder) createArgoTemplates(namespace string, _, firstItems, lastItems = b.addTaskToArgo(b.Workflow.getDag(), id, processing, firstItems, lastItems) template := &Template{Name: getArgoName(processing.GetName(), id)} fmt.Println("Creating template for", template.Name) - isReparted, url := b.isProcessingReparted(*processing) - template.CreateContainer(processing, b.Workflow.getDag(), isReparted, url) + isReparted, peerId := b.isProcessingReparted(*processing,id) + template.CreateContainer(processing, b.Workflow.getDag()) + if isReparted { + b.RemotePeers = append(b.RemotePeers, peerId) + } // get datacenter from the processing if processing.IsService { b.CreateService(id, processing) @@ -373,10 +376,15 @@ func getArgoName(raw_name string, component_id string) (formatedName string) { } // Verify if a processing resource is attached to another Compute than the one hosting -// the current Open Cloud instance. If true return the URL to contact the remote instance -// kube API -func (b *ArgoBuilder) isProcessingReparted(processing resources.ProcessingResource) (bool,string) { - processCreator := processing.CreatorID +// the current Open Cloud instance. If true return the peer ID to contact +func (b *ArgoBuilder) isProcessingReparted(processing resources.ProcessingResource, graphID string) (bool,string) { + computeAttached := b.retrieveProcessingCompute(graphID) + if computeAttached == nil { + logger.Error().Msg("No compute was found attached to processing " + processing.Name + " : " + processing.UUID ) + panic(0) + } + + // Creates an accessor srtictly for Peer Collection req := oclib.NewRequest(oclib.LibDataEnum(oclib.PEER),"","",nil,nil) if req == nil { @@ -384,79 +392,38 @@ func (b *ArgoBuilder) isProcessingReparted(processing resources.ProcessingResour return false, "" } - res := req.LoadOne(processCreator) + res := req.LoadOne(computeAttached.CreatorID) if res.Err != "" { fmt.Print("TODO : handle error when requesting PeerID") fmt.Print(res.Err) return false, "" } - peer := res.ToPeer() - if peer == nil { - fmt.Print("TODO : handle error when converting PeerID") - } + peer := *res.ToPeer() isNotReparted, _ := peer.IsMySelf() - if !isNotReparted { - // remoteCompute := b.retrieveProcessingCompute(processing) - // computeInstance := remoteCompute.GetSelectedInstance() - // if computeInstance == nil { - // fmt.Println("TODO: handle when retrieving instance") - // return false, "" - // } - - // instance := computeInstance.(*resources.ComputeResourceInstance) - // return true, instance.Source - //dataID == executionID - caller := tools.NewHTTPCaller( - map[tools.DataType]map[tools.METHOD]string{ - tools.ADMIRALTY_SOURCE: map[tools.METHOD]string{ - tools.POST : "/:id", - }, - tools.ADMIRALTY_KUBECONFIG: map[tools.METHOD]string{ - tools.POST: "/:id", - }, - tools.ADMIRALTY_SECRET: map[tools.METHOD]string{ - tools.POST: "/:id", - }, - tools.ADMIRALTY_TARGET: map[tools.METHOD]string{ - tools.POST: "/:id", - }, - tools.ADMIRALTY_NODES: map[tools.METHOD]string{ - tools.GET: "/id", - }, - }, - ) - - res, err := peer.LaunchPeerExecution(peer.UUID,"toto-5",tools.ADMIRALTY_SOURCE,tools.POST,nil,caller) - if err != nil { - fmt.Println("Error contacting remote peer") - fmt.Println(err) - panic(0) - } - fmt.Println(res) - // peer.LaunchPeerExecution(peer.UUID,"toto-5",ADMIRALTY_TOKEN,tools.GET,nil,caller) - } + fmt.Println("Result IsMySelf for ", peer.UUID ," : ", isNotReparted) - return false, "" + return !isNotReparted, peer.UUID } -func (b *ArgoBuilder) retrieveProcessingCompute(processing resources.ProcessingResource) *resources.ComputeResource { +func (b *ArgoBuilder) retrieveProcessingCompute(graphID string) *resources.ComputeResource { for _, link := range b.OriginWorkflow.Graph.Links { // If a link contains the id of the processing var oppositeId string - if link.Source.ID == processing.AbstractResource.UUID{ + if link.Source.ID == graphID{ oppositeId = link.Destination.ID - } else if(link.Destination.ID == processing.AbstractResource.UUID){ + } else if(link.Destination.ID == graphID){ oppositeId = link.Source.ID } - + fmt.Println("OppositeId : ", oppositeId) if oppositeId != "" { - isCompute, object := isCompute(oppositeId) - if !isCompute { + dt, res := b.OriginWorkflow.Graph.GetResource(oppositeId) + if dt == oclib.COMPUTE_RESOURCE { + return res.(*resources.ComputeResource) + } else { continue } - return object } } @@ -464,22 +431,14 @@ func (b *ArgoBuilder) retrieveProcessingCompute(processing resources.ProcessingR return nil } -func isCompute(resourceId string) (bool, *resources.ComputeResource) { - req := oclib.NewRequest(oclib.LibDataEnum(oclib.COMPUTE_RESOURCE),"","",nil,nil) - if req == nil { - fmt.Print("TODO : handle error when creating a NewRequest()") - return false, nil - } - res := req.LoadOne(resourceId) - - if res.Err != "" { - return false, nil - } - - compute := res.ToComputeResource() - if compute == nil { // Maybe we should add an Err returned by ToXXXXResource() - return false, nil - } - return true, compute +// Execute the last actions once the YAML file for the Argo Workflow is created +func (b *ArgoBuilder) CompleteBuild(executionsId string) error { + fmt.Println("DEV :: Completing build") + for _, peer := range b.RemotePeers { + fmt.Println("DEV :: Launching Admiralty Setup for ", peer) + setter := AdmiraltySetter{Id: executionsId} + setter.InitializeAdmiralty(conf.GetConfig().PeerID,peer) + } + return nil } \ No newline at end of file diff --git a/workflow_builder/graph.go b/workflow_builder/graph.go index 0f5168c..5716ee7 100644 --- a/workflow_builder/graph.go +++ b/workflow_builder/graph.go @@ -41,20 +41,20 @@ func (w *WorflowDB) getWorkflow(workflow_id string, peerID string) (workflow *wo return new_wf, nil } -func (w *WorflowDB) ExportToArgo(namespace string, timeout int) (string, int, error) { +func (w *WorflowDB) ExportToArgo(namespace string, timeout int) (*ArgoBuilder,string, int, error) { logger := oclib.GetLogger() fmt.Println("Exporting to Argo", w.Workflow) if len(w.Workflow.Name) == 0 || w.Workflow.Graph == nil { - return "", 0, fmt.Errorf("can't export a graph that has not been loaded yet") + return nil, "", 0, fmt.Errorf("can't export a graph that has not been loaded yet") } - argo_builder := ArgoBuilder{OriginWorkflow: w.Workflow, Timeout: timeout} - filename, stepMax, _, _, err := argo_builder.CreateDAG(namespace, true) + argoBuilder := ArgoBuilder{OriginWorkflow: w.Workflow, Timeout: timeout} + filename, stepMax, _, _, err := argoBuilder.CreateDAG(namespace, true) if err != nil { logger.Error().Msg("Could not create the argo file for " + w.Workflow.Name) - return "", 0, err + return nil, "", 0, err } - return filename, stepMax, nil + return &argoBuilder, filename, stepMax, nil } // TODO implement this function