package workflow import ( "bufio" "encoding/json" "errors" "fmt" "os" "strings" "time" "cloud.o-forge.io/core/oc-lib/models/collaborative_area/shallow_collaborative_area" "cloud.o-forge.io/core/oc-lib/models/common" "cloud.o-forge.io/core/oc-lib/models/common/pricing" "cloud.o-forge.io/core/oc-lib/models/peer" "cloud.o-forge.io/core/oc-lib/models/resources" "cloud.o-forge.io/core/oc-lib/models/utils" "cloud.o-forge.io/core/oc-lib/models/workflow/graph" "cloud.o-forge.io/core/oc-lib/tools" "github.com/google/uuid" ) /* * Workflow is a struct that represents a workflow * it defines the native workflow */ type Workflow struct { utils.AbstractObject // AbstractObject contains the basic fields of an object (id, name) resources.ResourceSet Graph *graph.Graph `bson:"graph,omitempty" json:"graph,omitempty"` // Graph UI & logic representation of the workflow ScheduleActive bool `json:"schedule_active" bson:"schedule_active"` // ScheduleActive is a flag that indicates if the schedule is active, if not the workflow is not scheduled and no execution or booking will be set // Schedule *WorkflowSchedule `bson:"schedule,omitempty" json:"schedule,omitempty"` // Schedule is the schedule of the workflow Shared []string `json:"shared,omitempty" bson:"shared,omitempty"` // Shared is the ID of the shared workflow // AbstractWorkflow contains the basic fields of a workflow } func (d *Workflow) GetAccessor(request *tools.APIRequest) utils.Accessor { return NewAccessor(request) // Create a new instance of the accessor } func (d *Workflow) GetResources(dt tools.DataType) []resources.ResourceInterface { itf := []resources.ResourceInterface{} switch dt { case tools.DATA_RESOURCE: for _, d := range d.DataResources { itf = append(itf, d) } return itf case tools.PROCESSING_RESOURCE: for _, d := range d.ProcessingResources { itf = append(itf, d) } return itf case tools.COMPUTE_RESOURCE: for _, d := range d.ComputeResources { itf = append(itf, d) } return itf case tools.WORKFLOW_RESOURCE: for _, d := range d.WorkflowResources { itf = append(itf, d) } return itf case tools.STORAGE_RESOURCE: for _, d := range d.StorageResources { itf = append(itf, d) } return itf } return itf } func (d *Workflow) ExtractFromPlantUML(plantUML *os.File, request *tools.APIRequest) (*Workflow, error) { if plantUML == nil { return d, errors.New("no file available to export") } defer plantUML.Close() d.Datas = []string{} d.Storages = []string{} d.Processings = []string{} d.Computes = []string{} d.Workflows = []string{} d.DataResources = []*resources.DataResource{} d.StorageResources = []*resources.StorageResource{} d.ProcessingResources = []*resources.ProcessingResource{} d.ComputeResources = []*resources.ComputeResource{} d.WorkflowResources = []*resources.WorkflowResource{} d.Graph = graph.NewGraph() resourceCatalog := map[string]func() resources.ResourceInterface{ "Processing": func() resources.ResourceInterface { return &resources.ProcessingResource{ AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.ProcessingInstance]{ Instances: []*resources.ProcessingInstance{}, }, } }, "Storage": func() resources.ResourceInterface { return &resources.StorageResource{ AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.StorageResourceInstance]{ Instances: []*resources.StorageResourceInstance{}, }, } }, "Data": func() resources.ResourceInterface { return &resources.DataResource{ AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.DataInstance]{ Instances: []*resources.DataInstance{}, }, } }, "ComputeUnit": func() resources.ResourceInterface { return &resources.ComputeResource{ AbstractInstanciatedResource: resources.AbstractInstanciatedResource[*resources.ComputeResourceInstance]{ Instances: []*resources.ComputeResourceInstance{}, }, } }, } graphVarName := map[string]*graph.GraphItem{} scanner := bufio.NewScanner(plantUML) for scanner.Scan() { line := scanner.Text() for n, new := range resourceCatalog { if strings.Contains(line, n+"(") && !strings.Contains(line, "!procedure") { // should exclude declaration of type. newRes := new() varName, graphItem, err := d.extractResourcePlantUML(line, newRes, n, request.PeerID) if err != nil { return d, err } graphVarName[varName] = graphItem continue } else if strings.Contains(line, n+"-->") { err := d.extractLink(line, graphVarName, "-->") if err != nil { fmt.Println(err) continue } } else if strings.Contains(line, n+"--") { err := d.extractLink(line, graphVarName, "--") if err != nil { fmt.Println(err) continue } } else if strings.Contains(line, n+"-") { err := d.extractLink(line, graphVarName, "-") if err != nil { fmt.Println(err) continue } } } } if err := scanner.Err(); err != nil { return d, err } d.generateResource(d.GetResources(tools.DATA_RESOURCE), request) d.generateResource(d.GetResources(tools.PROCESSING_RESOURCE), request) d.generateResource(d.GetResources(tools.STORAGE_RESOURCE), request) d.generateResource(d.GetResources(tools.COMPUTE_RESOURCE), request) d.generateResource(d.GetResources(tools.WORKFLOW_RESOURCE), request) return d, nil } func (d *Workflow) generateResource(datas []resources.ResourceInterface, request *tools.APIRequest) error { for _, d := range datas { access := d.GetAccessor(request) if _, code, err := access.LoadOne(d.GetID()); err != nil && code == 200 { continue } access.StoreOne(d) } return nil } func (d *Workflow) extractLink(line string, graphVarName map[string]*graph.GraphItem, pattern string) error { splitted := strings.Split(line, pattern) if len(splitted) < 2 { return errors.New("links elements not found") } if graphVarName[splitted[0]] != nil { return errors.New("links elements not found -> " + strings.Trim(splitted[0], " ")) } if graphVarName[splitted[1]] != nil { return errors.New("links elements not found -> " + strings.Trim(splitted[1], " ")) } link := &graph.GraphLink{ Source: graph.Position{ ID: graphVarName[splitted[0]].ID, X: 0, Y: 0, }, Destination: graph.Position{ ID: graphVarName[splitted[1]].ID, X: 0, Y: 0, }, } splittedComments := strings.Split(line, "'") if len(splittedComments) <= 1 { return errors.New("Can't deserialize Object, there's no commentary") } comment := strings.ReplaceAll(splittedComments[1], "'", "") // for now it's a json. json.Unmarshal([]byte(comment), link) d.Graph.Links = append(d.Graph.Links, *link) return nil } func (d *Workflow) extractResourcePlantUML(line string, resource resources.ResourceInterface, dataName string, peerID string) (string, *graph.GraphItem, error) { splittedFunc := strings.Split(line, "(") if len(splittedFunc) <= 1 { return "", nil, errors.New("Can't deserialize Object, there's no func") } splittedParams := strings.Split(splittedFunc[1], ",") if len(splittedFunc) <= 1 { return "", nil, errors.New("Can't deserialize Object, there's no params") } varName := splittedParams[0] splitted := strings.Split(splittedParams[1], "\"") if len(splitted) <= 1 { return "", nil, errors.New("Can't deserialize Object, there's no name") } resource.SetName(splitted[1]) splittedComments := strings.Split(line, "'") if len(splittedComments) <= 1 { return "", nil, errors.New("Can't deserialize Object, there's no commentary") } comment := strings.ReplaceAll(splittedComments[1], "'", "") // for now it's a json. instance := d.getNewInstance(dataName, splitted[1], peerID) if instance == nil { return "", nil, errors.New("No instance found.") } resource.AddInstances(instance) json.Unmarshal([]byte(comment), instance) // deserializer les instances... une instance doit par défaut avoir certaines valeurs d'accès. graphID := uuid.New() graphItem := &graph.GraphItem{ ID: graphID.String(), } graphItem = d.getNewGraphItem(dataName, graphItem, resource) d.Graph.Items[graphID.String()] = *graphItem return varName, graphItem, nil } func (d *Workflow) getNewGraphItem(dataName string, graphItem *graph.GraphItem, resource resources.ResourceInterface) *graph.GraphItem { switch dataName { case "Data": d.Datas = append(d.Datas, resource.GetID()) d.DataResources = append(d.DataResources, resource.(*resources.DataResource)) graphItem.Data = resource.(*resources.DataResource) case "Processing", "Event": d.Processings = append(d.Processings, resource.GetID()) d.ProcessingResources = append(d.ProcessingResources, resource.(*resources.ProcessingResource)) graphItem.Processing = resource.(*resources.ProcessingResource) case "Storage": d.Storages = append(d.Storages, resource.GetID()) d.StorageResources = append(d.StorageResources, resource.(*resources.StorageResource)) graphItem.Storage = resource.(*resources.StorageResource) case "ComputeUnit": d.Computes = append(d.Computes, resource.GetID()) d.ComputeResources = append(d.ComputeResources, resource.(*resources.ComputeResource)) graphItem.Compute = resource.(*resources.ComputeResource) default: return graphItem } return graphItem } func (d *Workflow) getNewInstance(dataName string, name string, peerID string) resources.ResourceInstanceITF { switch dataName { case "Data": return resources.NewDataInstance(name, peerID) case "Processing": return resources.NewProcessingInstance(name, peerID) case "Storage": return resources.NewStorageResourceInstance(name, peerID) case "ComputeUnit": return resources.NewComputeResourceInstance(name, peerID) default: return nil } } type Deps struct { Source string Dest string } func (w *Workflow) IsDependancy(id string) []Deps { dependancyOfIDs := []Deps{} for _, link := range w.Graph.Links { if _, ok := w.Graph.Items[link.Destination.ID]; !ok { continue } source := w.Graph.Items[link.Destination.ID].Processing if id == link.Source.ID && source != nil { dependancyOfIDs = append(dependancyOfIDs, Deps{Source: source.GetName(), Dest: link.Destination.ID}) } sourceWF := w.Graph.Items[link.Destination.ID].Workflow if id == link.Source.ID && sourceWF != nil { dependancyOfIDs = append(dependancyOfIDs, Deps{Source: sourceWF.GetName(), Dest: link.Destination.ID}) } } return dependancyOfIDs } func (w *Workflow) GetDependencies(id string) (dependencies []Deps) { for _, link := range w.Graph.Links { if _, ok := w.Graph.Items[link.Source.ID]; !ok { continue } source := w.Graph.Items[link.Source.ID].Processing if id == link.Destination.ID && source != nil { dependencies = append(dependencies, Deps{Source: source.GetName(), Dest: link.Source.ID}) continue } } return } func (w *Workflow) GetGraphItems(f func(item graph.GraphItem) bool) (list_datas []graph.GraphItem) { for _, item := range w.Graph.Items { if f(item) { list_datas = append(list_datas, item) } } return } func (w *Workflow) GetPricedItem( f func(item graph.GraphItem) bool, request *tools.APIRequest, buyingStrategy int, pricingStrategy int) map[string]pricing.PricedItemITF { list_datas := map[string]pricing.PricedItemITF{} for _, item := range w.Graph.Items { if f(item) { dt, res := item.GetResource() ord := res.ConvertToPricedResource(dt, request) list_datas[res.GetID()] = ord } } return list_datas } type Related struct { Node resources.ResourceInterface Links []graph.GraphLink } func (w *Workflow) GetByRelatedProcessing(processingID string, g func(item graph.GraphItem) bool) map[string]Related { related := map[string]Related{} for _, link := range w.Graph.Links { nodeID := link.Destination.ID var node resources.ResourceInterface if g(w.Graph.Items[link.Source.ID]) { item := w.Graph.Items[link.Source.ID] _, node = item.GetResource() } if node == nil && g(w.Graph.Items[link.Destination.ID]) { // if the source is not a storage, we consider that the destination is the storage nodeID = link.Source.ID item := w.Graph.Items[link.Destination.ID] // and the processing is the source _, node = item.GetResource() // we are looking for the storage as destination } if processingID == nodeID && node != nil { // if the storage is linked to the processing if _, ok := related[processingID]; !ok { related[processingID] = Related{} } rel := related[node.GetID()] rel.Node = node rel.Links = append(rel.Links, link) related[processingID] = rel } } return related } func (ao *Workflow) VerifyAuth(request *tools.APIRequest) bool { isAuthorized := false if len(ao.Shared) > 0 { for _, shared := range ao.Shared { shared, code, _ := shallow_collaborative_area.NewAccessor(request).LoadOne(shared) if code != 200 || shared == nil { isAuthorized = false } else { isAuthorized = shared.VerifyAuth(request) } } } return ao.AbstractObject.VerifyAuth(request) || isAuthorized } /* * CheckBooking is a function that checks the booking of the workflow on peers (even ourselves) */ func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) { // check if if wfa.Graph == nil { // no graph no booking return false, nil } accessor := (&resources.ComputeResource{}).GetAccessor(&tools.APIRequest{Caller: caller}) for _, link := range wfa.Graph.Links { if ok, compute_id := link.IsComputeLink(*wfa.Graph); ok { // check if the link is a link between a compute and a resource compute, code, _ := accessor.LoadOne(compute_id) if code != 200 { continue } // CHECK BOOKING ON PEER, compute could be a remote one peerID := compute.(*resources.ComputeResource).CreatorID if peerID == "" { return false, errors.New("no peer id") } // no peer id no booking, we need to know where to book _, err := (&peer.Peer{}).LaunchPeerExecution(peerID, compute_id, tools.BOOKING, tools.GET, nil, caller) if err != nil { return false, err } } } return true, nil } func (wf *Workflow) Planify(start time.Time, end *time.Time, request *tools.APIRequest) (float64, map[tools.DataType]map[string]pricing.PricedItemITF, *Workflow, error) { priceds := map[tools.DataType]map[string]pricing.PricedItemITF{} ps, priceds, err := plan[*resources.ProcessingResource](tools.PROCESSING_RESOURCE, wf, priceds, request, wf.Graph.IsProcessing, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) { return start.Add(time.Duration(wf.Graph.GetAverageTimeProcessingBeforeStart(0, res.GetID(), request)) * time.Second), priced.GetExplicitDurationInS() }, func(started time.Time, duration float64) *time.Time { s := started.Add(time.Duration(duration)) return &s }) if err != nil { return 0, priceds, nil, err } if _, priceds, err = plan[resources.ResourceInterface](tools.DATA_RESOURCE, wf, priceds, request, wf.Graph.IsData, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) { return start, 0 }, func(started time.Time, duration float64) *time.Time { return end }); err != nil { return 0, priceds, nil, err } for k, f := range map[tools.DataType]func(graph.GraphItem) bool{tools.STORAGE_RESOURCE: wf.Graph.IsStorage, tools.COMPUTE_RESOURCE: wf.Graph.IsCompute} { if _, priceds, err = plan[resources.ResourceInterface](k, wf, priceds, request, f, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) { nearestStart, longestDuration := wf.Graph.GetAverageTimeRelatedToProcessingActivity(start, ps, res, func(i graph.GraphItem) (r resources.ResourceInterface) { if f(i) { _, r = i.GetResource() } return r }, request) return start.Add(time.Duration(nearestStart) * time.Second), longestDuration }, func(started time.Time, duration float64) *time.Time { s := started.Add(time.Duration(duration)) return &s }); err != nil { return 0, priceds, nil, err } } longest := common.GetPlannerLongestTime(end, priceds, request) if _, priceds, err = plan[resources.ResourceInterface](tools.WORKFLOW_RESOURCE, wf, priceds, request, wf.Graph.IsWorkflow, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64) { start := start.Add(time.Duration(common.GetPlannerNearestStart(start, priceds, request)) * time.Second) longest := float64(-1) r, code, err := res.GetAccessor(request).LoadOne(res.GetID()) if code != 200 || err != nil { return start, longest } if neoLongest, _, _, err := r.(*Workflow).Planify(start, end, request); err != nil { return start, longest } else if neoLongest > longest { longest = neoLongest } return start.Add(time.Duration(common.GetPlannerNearestStart(start, priceds, request)) * time.Second), longest }, func(start time.Time, longest float64) *time.Time { s := start.Add(time.Duration(longest) * time.Second) return &s }); err != nil { return 0, priceds, nil, err } return longest, priceds, wf, nil } func plan[T resources.ResourceInterface]( dt tools.DataType, wf *Workflow, priceds map[tools.DataType]map[string]pricing.PricedItemITF, request *tools.APIRequest, f func(graph.GraphItem) bool, start func(resources.ResourceInterface, pricing.PricedItemITF) (time.Time, float64), end func(time.Time, float64) *time.Time) ([]T, map[tools.DataType]map[string]pricing.PricedItemITF, error) { resources := []T{} for _, item := range wf.GetGraphItems(f) { if priceds[dt] == nil { priceds[dt] = map[string]pricing.PricedItemITF{} } dt, realItem := item.GetResource() if realItem == nil { return resources, priceds, errors.New("could not load the processing resource") } priced := realItem.ConvertToPricedResource(dt, request) // Should be commented once the Pricing selection feature has been implemented, related to the commit d35ad440fa77763ec7f49ab34a85e47e75581b61 // if priced.SelectPricing() == nil { // return resources, priceds, errors.New("no pricings are selected... can't proceed") // } started, duration := start(realItem, priced) priced.SetLocationStart(started) if duration >= 0 { if e := end(started, duration); e != nil { priced.SetLocationEnd(*e) } } if e := end(started, priced.GetExplicitDurationInS()); e != nil { priced.SetLocationEnd(*e) } resources = append(resources, realItem.(T)) priceds[dt][item.ID] = priced } return resources, priceds, nil }