package workflow import ( "bufio" "encoding/json" "errors" "fmt" "mime/multipart" "strings" "time" "cloud.o-forge.io/core/oc-lib/models/booking" "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" ) type ConfigItem map[string]int func (c ConfigItem) Get(key string) *int { i := 0 if ins, ok := c[key]; ok { i = ins } return &i } /* * 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.NATIVE_TOOL: for _, d := range d.NativeTools { itf = append(itf, d) } return itf 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 multipart.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, "-->", false) if err != nil { fmt.Println(err) continue } } else if strings.Contains(line, n+"<--") { err := d.extractLink(line, graphVarName, "<--", true) if err != nil { fmt.Println(err) continue } } else if strings.Contains(line, n+"--") { err := d.extractLink(line, graphVarName, "--", false) if err != nil { fmt.Println(err) continue } } else if strings.Contains(line, n+"-") { err := d.extractLink(line, graphVarName, "-", false) 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, reverse bool) 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, }, } if reverse { tmp := link.Destination link.Destination = link.Source link.Source = tmp } 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": d.Processings = append(d.Processings, resource.GetID()) d.ProcessingResources = append(d.ProcessingResources, resource.(*resources.ProcessingResource)) graphItem.Processing = resource.(*resources.ProcessingResource) case "Event": access := resources.NewAccessor[*resources.NativeTool](tools.NATIVE_TOOL, &tools.APIRequest{ Admin: true, }, func() utils.DBObject { return &resources.NativeTool{} }) t, _, err := access.Search(nil, "WORKFLOW_EVENT", false) if err == nil && len(t) > 0 { d.NativeTool = append(d.NativeTool, t[0].GetID()) graphItem.NativeTool = t[0].(*resources.NativeTool) } 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) GetFirstItems() []graph.GraphItem { return w.GetGraphItems(func(item graph.GraphItem) bool { return len(w.GetDependencies(w.GetID())) == 0 }) } 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, instance int, partnership int, buying int, strategy int, bookingMode int, buyingStrategy int, pricingStrategy int) (map[string]pricing.PricedItemITF, error) { list_datas := map[string]pricing.PricedItemITF{} for _, item := range w.Graph.Items { if f(item) { dt, res := item.GetResource() ord, err := res.ConvertToPricedResource(dt, &instance, &partnership, &buying, &strategy, &bookingMode, request) if err != nil { return list_datas, err } list_datas[res.GetID()] = ord } } return list_datas, nil } 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 relID := node.GetID() rel := Related{} rel.Node = node rel.Links = append(rel.Links, link) related[relID] = rel } } return related } func (ao *Workflow) VerifyAuth(callName string, 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(callName, request) } } } return ao.AbstractObject.VerifyAuth(callName, 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, instances ConfigItem, partnerships ConfigItem, buyings ConfigItem, strategies ConfigItem, bookingMode int, request *tools.APIRequest) (bool, 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, instances, partnerships, buyings, strategies, bookingMode, wf, priceds, request, wf.Graph.IsProcessing, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64, error) { d, err := wf.Graph.GetAverageTimeProcessingBeforeStart(0, res.GetID(), *instances.Get(res.GetID()), *partnerships.Get(res.GetID()), *buyings.Get(res.GetID()), *strategies.Get(res.GetID()), bookingMode, request) if err != nil { return start, 0, err } return start.Add(time.Duration(d) * time.Second), priced.GetExplicitDurationInS(), nil }, func(started time.Time, duration float64) (*time.Time, error) { s := started.Add(time.Duration(duration)) return &s, nil }) if err != nil { return false, 0, priceds, nil, err } if _, priceds, err = plan[resources.ResourceInterface](tools.NATIVE_TOOL, instances, partnerships, buyings, strategies, bookingMode, wf, priceds, request, wf.Graph.IsNativeTool, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64, error) { return start, 0, nil }, func(started time.Time, duration float64) (*time.Time, error) { return end, nil }); err != nil { return false, 0, priceds, nil, err } if _, priceds, err = plan[resources.ResourceInterface](tools.DATA_RESOURCE, instances, partnerships, buyings, strategies, bookingMode, wf, priceds, request, wf.Graph.IsData, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64, error) { return start, 0, nil }, func(started time.Time, duration float64) (*time.Time, error) { return end, nil }); err != nil { return false, 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, instances, partnerships, buyings, strategies, bookingMode, wf, priceds, request, f, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64, error) { nearestStart, longestDuration, err := wf.Graph.GetAverageTimeRelatedToProcessingActivity(start, ps, res, func(i graph.GraphItem) (r resources.ResourceInterface) { if f(i) { _, r = i.GetResource() } return r }, *instances.Get(res.GetID()), *partnerships.Get(res.GetID()), *buyings.Get(res.GetID()), *strategies.Get(res.GetID()), bookingMode, request) if err != nil { return start, longestDuration, err } return start.Add(time.Duration(nearestStart) * time.Second), longestDuration, nil }, func(started time.Time, duration float64) (*time.Time, error) { s := started.Add(time.Duration(duration)) return &s, nil }); err != nil { return false, 0, priceds, nil, err } } longest := common.GetPlannerLongestTime(end, priceds, request) if _, priceds, err = plan[resources.ResourceInterface](tools.WORKFLOW_RESOURCE, instances, partnerships, buyings, strategies, bookingMode, wf, priceds, request, wf.Graph.IsWorkflow, func(res resources.ResourceInterface, priced pricing.PricedItemITF) (time.Time, float64, error) { 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, err } _, neoLongest, priceds2, _, err := r.(*Workflow).Planify(start, end, instances, partnerships, buyings, strategies, bookingMode, request) // should ... import priced if err != nil { return start, longest, err } else if neoLongest > longest { longest = neoLongest } for k, v := range priceds2 { if priceds[k] == nil { priceds[k] = map[string]pricing.PricedItemITF{} } for k2, v2 := range v { if priceds[k][k2] != nil { v2.AddQuantity(priceds[k][k2].GetQuantity()) } } } return start.Add(time.Duration(common.GetPlannerNearestStart(start, priceds, request)) * time.Second), longest, nil }, func(start time.Time, longest float64) (*time.Time, error) { s := start.Add(time.Duration(longest) * time.Second) return &s, nil }); err != nil { return false, 0, priceds, nil, err } isPreemptible := true for _, first := range wf.GetFirstItems() { _, res := first.GetResource() if res.GetBookingModes()[booking.PREEMPTED] == nil { isPreemptible = false break } } return isPreemptible, longest, priceds, wf, nil } // Returns a map of DataType (processing,computing,data,storage,worfklow) where each resource (identified by its UUID) // is mapped to the list of its items (different appearance) in the graph // ex: if the same Minio storage is represented by several nodes in the graph, in [tools.STORAGE_RESSOURCE] its UUID will be mapped to // the list of GraphItem ID that correspond to the ID of each node func (w *Workflow) GetItemsByResources() map[tools.DataType]map[string][]string { res := make(map[tools.DataType]map[string][]string) dtMethodMap := map[tools.DataType]func() []graph.GraphItem{ tools.STORAGE_RESOURCE: func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsStorage) }, tools.DATA_RESOURCE: func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsData) }, tools.COMPUTE_RESOURCE: func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsCompute) }, tools.PROCESSING_RESOURCE: func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsProcessing) }, tools.WORKFLOW_RESOURCE: func() []graph.GraphItem { return w.GetGraphItems(w.Graph.IsWorkflow) }, } for dt, meth := range dtMethodMap { res[dt] = make(map[string][]string) items := meth() for _, i := range items { _, r := i.GetResource() rId := r.GetID() res[dt][rId] = append(res[dt][rId], i.ID) } } return res } func plan[T resources.ResourceInterface]( dt tools.DataType, instances ConfigItem, partnerships ConfigItem, buyings ConfigItem, strategies ConfigItem, bookingMode int, 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, error), end func(time.Time, float64) (*time.Time, error)) ([]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, err := realItem.ConvertToPricedResource(dt, instances.Get(realItem.GetID()), partnerships.Get(realItem.GetID()), buyings.Get(realItem.GetID()), strategies.Get(realItem.GetID()), &bookingMode, request) if err != nil { return resources, priceds, err } // 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, err := start(realItem, priced) if err != nil { return resources, priceds, err } priced.SetLocationStart(started) if duration >= 0 { if e, err := end(started, duration); err == nil && e != nil { priced.SetLocationEnd(*e) } } if e, err := end(started, priced.GetExplicitDurationInS()); err != nil && e != nil { priced.SetLocationEnd(*e) } resources = append(resources, realItem.(T)) if priceds[dt][item.ID] != nil { priced.AddQuantity(priceds[dt][item.ID].GetQuantity()) } priceds[dt][item.ID] = priced } return resources, priceds, nil }