core/oc-lib
6
0
Fork 0
Code Issues 7 Pull Requests Packages Projects Releases Wiki Activity

lightest + clearest code

Browse Source
This commit is contained in:
mr
2024-11-28 16:49:41 +01:00
parent 8a11805fe8
commit 8d701e67d8
34 changed files with 444 additions and 1448 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
models/workflow/graph/graph.go Normal file
View File

@@ -0,0 +1,70 @@
package graph
import (
"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/tools"
)
// Graph is a struct that represents a graph
type Graph struct {
Zoom float64 `bson:"zoom" json:"zoom" default:"1"` // Zoom is the graphical zoom of the graph
Items map[string]GraphItem `bson:"items" json:"items" default:"{}" validate:"required"` // Items is the list of elements in the graph
Links []GraphLink `bson:"links" json:"links" default:"{}" validate:"required"` // Links is the list of links between elements in the graph
}
func (g *Graph) GetResource(id string) (string, utils.DBObject) {
if item, ok := g.Items[id]; ok {
if item.Data != nil {
return tools.DATA_RESOURCE.String(), item.Data
} else if item.Compute != nil {
return tools.COMPUTE_RESOURCE.String(), item.Compute
} else if item.Workflow != nil {
return tools.WORKFLOW_RESOURCE.String(), item.Workflow
} else if item.Processing != nil {
return tools.PROCESSING_RESOURCE.String(), item.Processing
} else if item.Storage != nil {
return tools.STORAGE_RESOURCE.String(), item.Storage
}
}
return "", nil
}
// GraphItem is a struct that represents an item in a graph
type GraphItem struct {
ID string `bson:"id" json:"id" validate:"required"` // ID is the unique identifier of the item
Width float64 `bson:"width" json:"width" validate:"required"` // Width is the graphical width of the item
Height float64 `bson:"height" json:"height" validate:"required"` // Height is the graphical height of the item
Position Position `bson:"position" json:"position" validate:"required"` // Position is the graphical position of the item
*resources.ItemResource // ItemResource is the resource of the item affected to the item
}
// GraphLink is a struct that represents a link between two items in a graph
type GraphLink struct {
Source Position `bson:"source" json:"source" validate:"required"` // Source is the source graphical position of the link
Destination Position `bson:"destination" json:"destination" validate:"required"` // Destination is the destination graphical position of the link
Style *GraphLinkStyle `bson:"style,omitempty" json:"style,omitempty"` // Style is the graphical style of the link
}
// GraphLinkStyle is a struct that represents the style of a link in a graph
type GraphLinkStyle struct {
Color int64 `bson:"color" json:"color"` // Color is the graphical color of the link (int description of a color, can be transpose as hex)
Stroke float64 `bson:"stroke" json:"stroke"` // Stroke is the graphical stroke of the link
Tension float64 `bson:"tension" json:"tension"` // Tension is the graphical tension of the link
HeadRadius float64 `bson:"head_radius" json:"head_radius"` // graphical pin radius
DashWidth float64 `bson:"dash_width" json:"dash_width"` // DashWidth is the graphical dash width of the link
DashSpace float64 `bson:"dash_space" json:"dash_space"` // DashSpace is the graphical dash space of the link
EndArrow Position `bson:"end_arrow" json:"end_arrow"` // EndArrow is the graphical end arrow of the link
StartArrow Position `bson:"start_arrow" json:"start_arrow"` // StartArrow is the graphical start arrow of the link
ArrowStyle int64 `bson:"arrow_style" json:"arrow_style"` // ArrowStyle is the graphical arrow style of the link (enum foundable in UI)
ArrowDirection int64 `bson:"arrow_direction" json:"arrow_direction"` // ArrowDirection is the graphical arrow direction of the link (enum foundable in UI)
StartArrowWidth float64 `bson:"start_arrow_width" json:"start_arrow_width"` // StartArrowWidth is the graphical start arrow width of the link
EndArrowWidth float64 `bson:"end_arrow_width" json:"end_arrow_width"` // EndArrowWidth is the graphical end arrow width of the link
}
// Position is a struct that represents a graphical position
type Position struct {
ID string `json:"id" bson:"id"` // ID reprents ItemID (optionnal), TODO: rename to ItemID
X float64 `json:"x" bson:"x" validate:"required"` // X is the graphical x position
Y float64 `json:"y" bson:"y" validate:"required"` // Y is the graphical y position
}

16
models/workflow/workflow.go
View File

@@ -5,10 +5,8 @@ import (
"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/resources/compute"
"cloud.o-forge.io/core/oc-lib/models/resources/storage"
"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
"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"
)
@@ -35,8 +33,8 @@ func (w *AbstractWorkflow) GetWorkflows() (list_computings []graph.GraphItem) {
return
}
func (w *AbstractWorkflow) GetComputeByRelatedProcessing(processingID string) []*compute.ComputeResource {
storages := []*compute.ComputeResource{}
func (w *AbstractWorkflow) GetComputeByRelatedProcessing(processingID string) []*resources.ComputeResource {
storages := []*resources.ComputeResource{}
for _, link := range w.Graph.Links {
nodeID := link.Destination.ID // we considers that the processing is the destination
node := w.Graph.Items[link.Source.ID].Compute // we are looking for the storage as source
@@ -51,8 +49,8 @@ func (w *AbstractWorkflow) GetComputeByRelatedProcessing(processingID string) []
return storages
}
func (w *AbstractWorkflow) GetStoragesByRelatedProcessing(processingID string) []*storage.StorageResource {
storages := []*storage.StorageResource{}
func (w *AbstractWorkflow) GetStoragesByRelatedProcessing(processingID string) []*resources.StorageResource {
storages := []*resources.StorageResource{}
for _, link := range w.Graph.Links {
nodeID := link.Destination.ID // we considers that the processing is the destination
node := w.Graph.Items[link.Source.ID].Storage // we are looking for the storage as source
@@ -107,7 +105,7 @@ func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
if wfa.Graph == nil { // no graph no booking
return false, nil
}
accessor := (&compute.ComputeResource{}).GetAccessor("", []string{}, caller)
accessor := (&resources.ComputeResource{}).GetAccessor("", []string{}, caller)
for _, link := range wfa.Graph.Links {
if ok, dc_id := wfa.isDCLink(link); ok { // check if the link is a link between a compute and a resource
dc, code, _ := accessor.LoadOne(dc_id)
@@ -115,7 +113,7 @@ func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
continue
}
// CHECK BOOKING ON PEER, compute could be a remote one
peerID := dc.(*compute.ComputeResource).PeerID
peerID := dc.(*resources.ComputeResource).PeerID
if peerID == "" {
return false, errors.New("no peer id")
} // no peer id no booking, we need to know where to book

85
models/workflow/workflow_mongo_accessor.go
View File

@@ -42,7 +42,7 @@ func New(t tools.DataType, peerID string, groups []string, caller *tools.HTTPCal
Caller: caller,
PeerID: peerID,
Groups: groups, // Set the caller
Type: t.String(),
Type: t,
},
}
}
@@ -157,7 +157,7 @@ func (wfa *workflowMongoAccessor) book(id string, realData *Workflow, execs []*w
continue
}
// CHECK BOOKING
peerID := dc.(*compute.ComputeResource).PeerID
peerID := dc.(*resources.ComputeResource).PeerID
if peerID == "" { // no peer id no booking
continue
}
@@ -319,7 +319,7 @@ func (wfa *workflowMongoAccessor) execute(workflow *Workflow, delete bool, activ
filters := &dbs.Filters{
Or: map[string][]dbs.Filter{ // filter by standard workspace name attached to a workflow
"abstractobject.name": {{dbs.LIKE.String(), workflow.Name + "_workspace"}},
"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: workflow.Name + "_workspace"}},
},
}
resource, _, err := wfa.workspaceAccessor.Search(filters, "")
@@ -355,64 +355,25 @@ func (wfa *workflowMongoAccessor) execute(workflow *Workflow, delete bool, activ
}
}
// LoadOne loads a workflow from the database
func (wfa *workflowMongoAccessor) LoadOne(id string) (utils.DBObject, int, error) {
var workflow Workflow
res_mongo, code, err := mongo.MONGOService.LoadOne(id, wfa.GetType())
if err != nil {
wfa.Logger.Error().Msg("Could not retrieve " + id + " from db. Error: " + err.Error())
return nil, code, err
}
res_mongo.Decode(&workflow)
if workflow.ScheduleActive && workflow.Schedule != nil { // if the workflow is scheduled, update the executions
now := time.Now().UTC()
if (workflow.Schedule.End != nil && now.After(*workflow.Schedule.End)) || (workflow.Schedule.End == nil && workflow.Schedule.Start != nil && now.After(*workflow.Schedule.Start)) { // if the start date is passed, then you can book
workflow.ScheduleActive = false
wfa.GenericRawUpdateOne(&workflow, id, wfa)
} // if the start date is passed, update the executions
}
wfa.execute(&workflow, false, true) // if no workspace is attached to the workflow, create it
return &workflow, 200, nil
}
// LoadAll loads all the workflows from the database
func (wfa workflowMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int, error) {
objs := []utils.ShallowDBObject{}
res_mongo, code, err := mongo.MONGOService.LoadAll(wfa.GetType())
if err != nil {
wfa.Logger.Error().Msg("Could not retrieve any from db. Error: " + err.Error())
return nil, code, err
}
var results []Workflow
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
for _, r := range results {
objs = append(objs, &r.AbstractObject) // only AbstractObject fields !
}
return objs, 200, nil
}
func (wfa *workflowMongoAccessor) Search(filters *dbs.Filters, search string) ([]utils.ShallowDBObject, int, error) {
objs := []utils.ShallowDBObject{}
if (filters == nil || len(filters.And) == 0 || len(filters.Or) == 0) && search != "" {
filters = &dbs.Filters{
Or: map[string][]dbs.Filter{ // filter by name if no filters are provided
"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: search}},
},
func (a *workflowMongoAccessor) LoadOne(id string) (utils.DBObject, int, error) {
return utils.GenericLoadOne[*Workflow](id, func(d utils.DBObject) (utils.DBObject, int, error) {
w := d.(*Workflow)
if w.ScheduleActive && w.Schedule != nil { // if the workflow is scheduled, update the executions
now := time.Now().UTC()
if (w.Schedule.End != nil && now.After(*w.Schedule.End)) || (w.Schedule.End == nil && w.Schedule.Start != nil && now.After(*w.Schedule.Start)) { // if the start date is passed, then you can book
w.ScheduleActive = false
a.GenericRawUpdateOne(d, id, a)
} // if the start date is passed, update the executions
}
}
res_mongo, code, err := mongo.MONGOService.Search(filters, wfa.GetType())
if err != nil {
wfa.Logger.Error().Msg("Could not store to db. Error: " + err.Error())
return nil, code, err
}
var results []Workflow
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
for _, r := range results {
objs = append(objs, &r)
}
return objs, 200, nil
a.execute(w, false, true) // if no workspace is attached to the workflow, create it
return d, 200, nil
}, a)
}
func (a *workflowMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int, error) {
return utils.GenericLoadAll[*Workflow](func(d utils.DBObject) utils.ShallowDBObject { return &d.(*Workflow).AbstractObject }, a)
}
func (a *workflowMongoAccessor) Search(filters *dbs.Filters, search string) ([]utils.ShallowDBObject, int, error) {
return utils.GenericSearch[*Workflow](filters, search, (&Workflow{}).GetObjectFilters(search), func(d utils.DBObject) utils.ShallowDBObject { return d }, a)
}