Merge branch 'main' of https://cloud.o-forge.io/core/oc-lib

Merged Main OCLIB

models/bill/bill.go

@@ -2,6 +2,7 @@ package bill
 
 import (
 	"encoding/json"
+	"fmt"
 	"sync"
 	"time"
 
@@ -184,9 +185,18 @@ type PeerItemOrder struct {
 }
 
 func (d *PeerItemOrder) GetPrice(request *tools.APIRequest) (float64, error) {
+	/////////// Temporary in order to allow GenerateOrder to complete while billing is still WIP
+	if d.Purchase == nil {
+		return 0, nil
+	}
+	///////////
 	var priced *resources.PricedResource
 	b, _ := json.Marshal(d.Item)
-	json.Unmarshal(b, priced)
+	err := json.Unmarshal(b, priced)
+	if err != nil {
+		fmt.Println(err)
+		return 0, err
+	}
 	accessor := purchase_resource.NewAccessor(request)
 	search, code, _ := accessor.Search(&dbs.Filters{
 		And: map[string][]dbs.Filter{

models/booking/booking.go

@@ -22,7 +22,7 @@ type Booking struct {
 	ExecutionMetrics map[string][]models.MetricsSnapshot       `json:"metrics,omitempty" bson:"metrics,omitempty"`
 
 	ExecutionsID      string             `json:"executions_id,omitempty" bson:"executions_id,omitempty" validate:"required"` // ExecutionsID is the ID of the executions
-	DestPeerID        string             `json:"dest_peer_id,omitempty"`                                                     // DestPeerID is the ID of the destination peer
+	DestPeerID        string             `json:"dest_peer_id,omitempty" bson:"dest_peer_id,omitempty"`                                                     // DestPeerID is the ID of the destination peer
 	WorkflowID        string             `json:"workflow_id,omitempty" bson:"workflow_id,omitempty"`                         // WorkflowID is the ID of the workflow
 	ExecutionID       string             `json:"execution_id,omitempty" bson:"execution_id,omitempty" validate:"required"`
 	State             enum.BookingStatus `json:"state,omitempty" bson:"state,omitempty" validate:"required"`                             // State is the state of the booking

models/resources/purchase_resource/purchase_resource.go

@@ -9,7 +9,7 @@ import (
 
 type PurchaseResource struct {
 	utils.AbstractObject
-	DestPeerID   string
+	DestPeerID   string				 	`json:"dest_peer_id" bson:"dest_peer_id"`
 	PricedItem   map[string]interface{} `json:"priced_item,omitempty" bson:"priced_item,omitempty" validate:"required"`
 	ExecutionsID string                 `json:"executions_id,omitempty" bson:"executions_id,omitempty" validate:"required"` // ExecutionsID is the ID of the executions
 	EndDate      *time.Time             `json:"end_buying_date,omitempty" bson:"end_buying_date,omitempty"`

models/workflow/workflow.go

@@ -367,13 +367,11 @@ func (w *Workflow) GetByRelatedProcessing(processingID string, g func(item graph
 			_, 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()]
+			relID := node.GetID()
+			rel := Related{}
 			rel.Node = node
 			rel.Links = append(rel.Links, link)
-			related[processingID] = rel
+			related[relID] = rel
 		}
 	}
 	return related
@@ -484,6 +482,33 @@ func (wf *Workflow) Planify(start time.Time, end *time.Time, request *tools.APIR
 	return 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, 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) {

models/workflow_execution/workflow_scheduler.go

@@ -142,7 +142,7 @@ func getBooking(b *booking.Booking, request *tools.APIRequest, errCh chan error,
 	_, err = (&peer.Peer{}).LaunchPeerExecution(b.DestPeerID, b.ResourceID, tools.BOOKING, tools.GET, nil, &c)
 
 	if err != nil {
-		errCh <- err
+		errCh <- fmt.Errorf("error on " + b.DestPeerID + err.Error())
 		return
 	}