651 lines
23 KiB
Go
651 lines
23 KiB
Go
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
|
|
}
|