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

massive draft for payment process (UNCOMPLETE)

Browse Source
This commit is contained in:
mr
2024-12-12 16:25:47 +01:00
parent fbbce7817b
commit 02d1e93c78
55 changed files with 3018 additions and 1177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
models/workflow/graph/graph.go
View File

@@ -1,8 +1,10 @@
package graph
import (
"time"
"cloud.o-forge.io/core/oc-lib/models/common/pricing"
"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"
)
@@ -13,7 +15,84 @@ type Graph struct {
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) {
func (g *Graph) GetAverageTimeRelatedToProcessingActivity(start time.Time, processings []*resources.CustomizedProcessingResource, resource resources.ShallowResourceInterface, f func(GraphItem) resources.ShallowResourceInterface) (float64, float64) {
nearestStart := float64(10000000000)
oneIsInfinite := false
longestDuration := float64(0)
for _, link := range g.Links {
for _, processing := range processings {
var source string // source is the source of the link
if link.Destination.ID == processing.GetID() && f(g.Items[link.Source.ID]) != nil && f(g.Items[link.Source.ID]).GetID() == resource.GetID() { // if the destination is the processing and the source is not a compute
source = link.Source.ID
} else if link.Source.ID == processing.GetID() && f(g.Items[link.Source.ID]) != nil && f(g.Items[link.Source.ID]).GetID() == resource.GetID() { // if the source is the processing and the destination is not a compute
source = link.Destination.ID
}
if source != "" {
if processing.UsageStart != nil {
near := float64(processing.UsageStart.Sub(start).Seconds())
if near < nearestStart {
nearestStart = near
}
}
if processing.UsageEnd != nil {
duration := float64(processing.UsageEnd.Sub(*processing.UsageStart).Seconds())
if longestDuration < duration {
longestDuration = duration
}
} else {
oneIsInfinite = true
}
}
}
}
if oneIsInfinite {
return nearestStart, -1
}
return nearestStart, longestDuration
}
func (g *Graph) SetItemStartUsage(graphItemID string, start time.Time) {
g.Items[graphItemID].SetItemStartUsage(start)
}
func (g *Graph) SetItemEndUsage(graphItemID string, end time.Time) {
g.Items[graphItemID].SetItemEndUsage(end)
}
/*
* GetAverageTimeBeforeStart is a function that returns the average time before the start of a processing
*/
func (g *Graph) GetAverageTimeProcessingBeforeStart(average float64, processingID string) float64 {
currents := []float64{} // list of current time
for _, link := range g.Links { // for each link
var source string // source is the source of the link
if link.Destination.ID == processingID && g.Items[link.Source.ID].Processing == nil { // if the destination is the processing and the source is not a compute
source = link.Source.ID
} else if link.Source.ID == processingID && g.Items[link.Source.ID].Processing == nil { // if the source is the processing and the destination is not a compute
source = link.Destination.ID
}
if source == "" { // if source is empty, continue
continue
}
_, item := g.GetResource(source) // get the resource of the source
current := item.GetExplicitDurationInS() // get the explicit duration of the item
if current < 0 { // if current is negative, its means that duration of a before could be infinite continue
return current
}
current += g.GetAverageTimeProcessingBeforeStart(current, source) // get the average time before start of the source
currents = append(currents, current) // append the current to the currents
}
var max float64 // get the max time to wait dependancies to finish
for _, current := range currents {
if current > max {
max = current
}
}
return max
}
func (g *Graph) GetResource(id string) (string, resources.ShallowResourceInterface) {
if item, ok := g.Items[id]; ok {
if item.Data != nil {
return tools.DATA_RESOURCE.String(), item.Data
@@ -32,11 +111,41 @@ func (g *Graph) GetResource(id string) (string, utils.DBObject) {
// 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
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.ItemExploitedResource // ItemResource is the resource of the item affected to the item
}
func (g *GraphItem) GetResource() resources.ShallowResourceInterface {
if g.Data != nil {
return g.Data
} else if g.Compute != nil {
return g.Compute
} else if g.Workflow != nil {
return g.Workflow
} else if g.Processing != nil {
return g.Processing
} else if g.Storage != nil {
return g.Storage
}
return nil
}
func (g *GraphItem) GetPricedItem() pricing.PricedItemITF {
if g.Data != nil {
return g.Data
} else if g.Compute != nil {
return g.Compute
} else if g.Workflow != nil {
return g.Workflow
} else if g.Processing != nil {
return g.Processing
} else if g.Storage != nil {
return g.Storage
}
return nil
}
// GraphLink is a struct that represents a link between two items in a graph
@@ -46,6 +155,20 @@ type GraphLink struct {
Style *GraphLinkStyle `bson:"style,omitempty" json:"style,omitempty"` // Style is the graphical style of the link
}
// tool function to check if a link is a link between a compute and a resource
func (l *GraphLink) IsComputeLink(g Graph) (bool, string) {
if g.Items == nil {
return false, ""
}
if d, ok := g.Items[l.Source.ID]; ok && d.Compute != nil {
return true, d.Compute.UUID
}
if d, ok := g.Items[l.Destination.ID]; ok && d.Compute != nil {
return true, d.Compute.UUID
}
return false, ""
}
// 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)
@@ -64,7 +187,7 @@ type GraphLinkStyle struct {
// 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
ID string `json:"id" bson:"id"` // ID reprents ItemID (optionnal)
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
}

164
models/workflow/workflow.go
View File

@@ -2,8 +2,10 @@ package workflow
import (
"errors"
"time"
"cloud.o-forge.io/core/oc-lib/models/collaborative_area/shallow_collaborative_area"
"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"
@@ -19,29 +21,61 @@ import (
*/
type AbstractWorkflow struct {
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
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
}
func (w *AbstractWorkflow) GetWorkflows() (list_computings []graph.GraphItem) {
func (d *Workflow) GetAccessor(request *tools.APIRequest) utils.Accessor {
return NewAccessor(request) // Create a new instance of the accessor
}
func (w *AbstractWorkflow) GetGraphItems(f func(item graph.GraphItem) bool) (list_datas []graph.GraphItem) {
for _, item := range w.Graph.Items {
if item.Workflow != nil {
list_computings = append(list_computings, item)
if f(item) {
list_datas = append(list_datas, item)
}
}
return
}
func (w *AbstractWorkflow) GetComputeByRelatedProcessing(processingID string) []*resources.ComputeResource {
storages := []*resources.ComputeResource{}
func (w *AbstractWorkflow) GetResources(f func(item graph.GraphItem) bool) map[string]resources.ShallowResourceInterface {
list_datas := map[string]resources.ShallowResourceInterface{}
for _, item := range w.Graph.Items {
if f(item) {
res := item.GetResource()
list_datas[res.GetID()] = res
}
}
return list_datas
}
func (w *AbstractWorkflow) GetPricedItem(f func(item graph.GraphItem) bool) map[string]pricing.PricedItemITF {
list_datas := map[string]pricing.PricedItemITF{}
for _, item := range w.Graph.Items {
if f(item) {
res := item.GetResource()
ord := item.GetPricedItem()
list_datas[res.GetID()] = ord
}
}
return list_datas
}
func (w *AbstractWorkflow) GetByRelatedProcessing(processingID string, g func(item graph.GraphItem) bool) []resources.ShallowResourceInterface {
storages := []resources.ShallowResourceInterface{}
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
if node == nil { // if the source is not a storage, we consider that the destination is the storage
nodeID = link.Source.ID // and the processing is the source
node = w.Graph.Items[link.Destination.ID].Compute // we are looking for the storage as destination
nodeID := link.Destination.ID
var node resources.ShallowResourceInterface
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
storages = append(storages, node)
@@ -50,43 +84,24 @@ func (w *AbstractWorkflow) GetComputeByRelatedProcessing(processingID string) []
return storages
}
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
if node == nil { // if the source is not a storage, we consider that the destination is the storage
nodeID = link.Source.ID // and the processing is the source
node = w.Graph.Items[link.Destination.ID].Storage // we are looking for the storage as destination
}
if processingID == nodeID && node != nil { // if the storage is linked to the processing
storages = append(storages, node)
}
}
return storages
func (wf *AbstractWorkflow) IsProcessing(item graph.GraphItem) bool {
return item.Processing != nil
}
func (w *AbstractWorkflow) GetProcessings() (list_computings []graph.GraphItem) {
for _, item := range w.Graph.Items {
if item.Processing != nil {
list_computings = append(list_computings, item)
}
}
return
func (wf *AbstractWorkflow) IsCompute(item graph.GraphItem) bool {
return item.Compute != nil
}
// tool function to check if a link is a link between a compute and a resource
func (w *AbstractWorkflow) isDCLink(link graph.GraphLink) (bool, string) {
if w.Graph == nil || w.Graph.Items == nil {
return false, ""
}
if d, ok := w.Graph.Items[link.Source.ID]; ok && d.Compute != nil {
return true, d.Compute.UUID
}
if d, ok := w.Graph.Items[link.Destination.ID]; ok && d.Compute != nil {
return true, d.Compute.UUID
}
return false, ""
func (wf *AbstractWorkflow) IsData(item graph.GraphItem) bool {
return item.Data != nil
}
func (wf *AbstractWorkflow) IsStorage(item graph.GraphItem) bool {
return item.Storage != nil
}
func (wf *AbstractWorkflow) IsWorkflow(item graph.GraphItem) bool {
return item.Workflow != nil
}
/*
@@ -98,18 +113,51 @@ type Workflow struct {
AbstractWorkflow // AbstractWorkflow contains the basic fields of a workflow
}
func (ao *Workflow) VerifyAuth(username string, peerID string, groups []string) bool {
func (w *Workflow) GetNearestStart(start time.Time) float64 {
near := float64(10000000000)
for _, item := range w.Graph.Items {
if item.GetResource().GetLocationStart() == nil {
continue
}
newS := item.GetResource().GetLocationStart()
if newS.Sub(start).Seconds() < near {
near = newS.Sub(start).Seconds()
}
// get the nearest start from start var
}
return near
}
func (w *Workflow) GetLongestTime(end *time.Time) float64 {
if end == nil {
return -1
}
longestTime := float64(0)
for _, item := range w.GetGraphItems(w.IsProcessing) {
if item.GetResource().GetLocationEnd() == nil {
continue
}
newS := item.GetResource().GetLocationEnd()
if longestTime < newS.Sub(*end).Seconds() {
longestTime = newS.Sub(*end).Seconds()
}
// get the nearest start from start var
}
return longestTime
}
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.New(tools.COLLABORATIVE_AREA, username, peerID, groups, nil).LoadOne(shared)
shared, code, _ := shallow_collaborative_area.NewAccessor(request).LoadOne(shared)
if code != 200 || shared == nil {
isAuthorized = false
}
isAuthorized = shared.VerifyAuth(username, peerID, groups)
isAuthorized = shared.VerifyAuth(request)
}
}
return ao.AbstractObject.VerifyAuth(username, peerID, groups) || isAuthorized
return ao.AbstractObject.VerifyAuth(request) || isAuthorized
}
/*
@@ -120,19 +168,19 @@ func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
if wfa.Graph == nil { // no graph no booking
return false, nil
}
accessor := (&resources.ComputeResource{}).GetAccessor("", "", []string{}, caller)
accessor := (&resources.ComputeResource{}).GetAccessor(&tools.APIRequest{Caller: 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)
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 := dc.(*resources.ComputeResource).PeerID
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, dc_id, tools.BOOKING, tools.GET, nil, caller)
_, err := (&peer.Peer{}).LaunchPeerExecution(peerID, compute_id, tools.BOOKING, tools.GET, nil, caller)
if err != nil {
return false, err
}
@@ -140,7 +188,3 @@ func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
}
return true, nil
}
func (d *Workflow) GetAccessor(username string, peerID string, groups []string, caller *tools.HTTPCaller) utils.Accessor {
return New(tools.WORKFLOW, username, peerID, groups, caller) // Create a new instance of the accessor
}

4
models/workflow/workflow_history_mongo_accessor.go
View File

@@ -8,8 +8,8 @@ import (
type WorkflowHistory struct{ Workflow }
func (d *WorkflowHistory) GetAccessor(username string, peerID string, groups []string, caller *tools.HTTPCaller) utils.Accessor {
return New(tools.WORKSPACE_HISTORY, username, peerID, groups, caller) // Create a new instance of the accessor
func (d *WorkflowHistory) GetAccessor(request tools.APIRequest) utils.Accessor {
return NewAccessorHistory(request) // Create a new instance of the accessor
}
func (r *WorkflowHistory) GenerateID() {
r.UUID = uuid.New().String()

264
models/workflow/workflow_mongo_accessor.go
View File

@@ -1,183 +1,55 @@
package workflow
import (
"errors"
"fmt"
"slices"
"strings"
"time"
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/dbs/mongo"
"cloud.o-forge.io/core/oc-lib/logs"
"cloud.o-forge.io/core/oc-lib/models/collaborative_area/shallow_collaborative_area"
"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_execution"
"cloud.o-forge.io/core/oc-lib/models/workspace"
"cloud.o-forge.io/core/oc-lib/tools"
cron "github.com/robfig/cron"
)
type workflowMongoAccessor struct {
utils.AbstractAccessor // AbstractAccessor contains the basic fields of an accessor (model, caller)
utils.AbstractAccessor // AbstractAccessor contains the basic fields of an accessor (model, caller)
computeResourceAccessor utils.Accessor
collaborativeAreaAccessor utils.Accessor
executionAccessor utils.Accessor
workspaceAccessor utils.Accessor
}
func NewAccessorHistory(request *tools.APIRequest) *workflowMongoAccessor {
return new(tools.WORKFLOW_HISTORY, request)
}
func NewAccessor(request *tools.APIRequest) *workflowMongoAccessor {
return new(tools.WORKFLOW, request)
}
// New creates a new instance of the workflowMongoAccessor
func New(t tools.DataType, username string, peerID string, groups []string, caller *tools.HTTPCaller) *workflowMongoAccessor {
func new(t tools.DataType, request *tools.APIRequest) *workflowMongoAccessor {
return &workflowMongoAccessor{
computeResourceAccessor: (&resources.ComputeResource{}).GetAccessor(username, peerID, groups, nil),
collaborativeAreaAccessor: (&shallow_collaborative_area.ShallowCollaborativeArea{}).GetAccessor(username, peerID, groups, nil),
executionAccessor: (&workflow_execution.WorkflowExecution{}).GetAccessor(username, peerID, groups, nil),
workspaceAccessor: (&workspace.Workspace{}).GetAccessor(username, peerID, groups, nil),
computeResourceAccessor: (&resources.ComputeResource{}).GetAccessor(request),
collaborativeAreaAccessor: (&shallow_collaborative_area.ShallowCollaborativeArea{}).GetAccessor(request),
workspaceAccessor: (&workspace.Workspace{}).GetAccessor(request),
AbstractAccessor: utils.AbstractAccessor{
Logger: logs.CreateLogger(t.String()), // Create a logger with the data type
Caller: caller,
PeerID: peerID,
User: username,
Groups: groups, // Set the caller
Type: t,
Logger: logs.CreateLogger(t.String()), // Create a logger with the data type
Request: request,
Type: t,
},
}
}
/*
* THERE IS A LOT IN THIS FILE SHOULD BE AWARE OF THE COMMENTS
*/
/*
* getExecutions is a function that returns the executions of a workflow
* it returns an array of workflow_execution.WorkflowExecution
*/
func (a *workflowMongoAccessor) getExecutions(id string, data *Workflow) ([]*workflow_execution.WorkflowExecution, error) {
workflows_execution := []*workflow_execution.WorkflowExecution{}
if data.Schedule != nil { // only set execution on a scheduled workflow
if data.Schedule.Start == nil { // if no start date, return an error
return workflows_execution, errors.New("should get a start date on the scheduler.")
}
if data.Schedule.End != nil && data.Schedule.End.IsZero() { // if end date is zero, set it to nil
data.Schedule.End = nil
}
if len(data.Schedule.Cron) > 0 { // if cron is set then end date should be set
if data.Schedule.End == nil {
return workflows_execution, errors.New("a cron task should have an end date.")
}
cronStr := strings.Split(data.Schedule.Cron, " ") // split the cron string to treat it
if len(cronStr) < 6 { // if the cron string is less than 6 fields, return an error because format is : ss mm hh dd MM dw (6 fields)
return nil, errors.New("Bad cron message: " + data.Schedule.Cron + ". Should be at least ss mm hh dd MM dw")
}
subCron := strings.Join(cronStr[:6], " ")
// cron should be parsed as ss mm hh dd MM dw t (min 6 fields)
specParser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) // create a new cron parser
sched, err := specParser.Parse(subCron) // parse the cron string
if err != nil {
return workflows_execution, errors.New("Bad cron message: " + err.Error())
}
// loop through the cron schedule to set the executions
for s := sched.Next(*data.Schedule.Start); !s.IsZero() && s.Before(*data.Schedule.End); s = sched.Next(s) {
obj := &workflow_execution.WorkflowExecution{
AbstractObject: utils.AbstractObject{
Name: data.Schedule.Name, // set the name of the execution
},
ExecDate: &s, // set the execution date
EndDate: data.Schedule.End, // set the end date
State: 1, // set the state to 1 (scheduled)
WorkflowID: id, // set the workflow id dependancy of the execution
}
workflows_execution = append(workflows_execution, obj) // append the execution to the array
}
} else { // if no cron, set the execution to the start date
obj := &workflow_execution.WorkflowExecution{ // create a new execution
AbstractObject: utils.AbstractObject{
Name: data.Schedule.Name,
},
ExecDate: data.Schedule.Start,
EndDate: data.Schedule.End,
State: 1,
WorkflowID: id,
}
workflows_execution = append(workflows_execution, obj) // append the execution to the array
}
}
return workflows_execution, nil
}
// DeleteOne deletes a workflow from the database, delete depending executions and bookings
func (a *workflowMongoAccessor) DeleteOne(id string) (utils.DBObject, int, error) {
a.execution(id, &Workflow{
AbstractWorkflow: AbstractWorkflow{ScheduleActive: false},
}, true) // delete the executions
res, code, err := utils.GenericDeleteOne(id, a)
if res != nil && code == 200 {
a.execute(res.(*Workflow), true, false) // up to date the workspace for the workflow
a.share(res.(*Workflow), true, a.Caller)
a.share(res.(*Workflow), true, a.GetCaller())
}
return res, code, err
}
/*
* book is a function that books a workflow on the peers
* it takes the workflow id, the real data and the executions
* it returns an error if the booking fails
*/
func (a *workflowMongoAccessor) book(id string, realData *Workflow, execs []*workflow_execution.WorkflowExecution) error {
if a.Caller == nil || a.Caller.URLS == nil || a.Caller.URLS[tools.BOOKING] == nil {
return errors.New("no caller defined")
}
methods := a.Caller.URLS[tools.BOOKING]
if _, ok := methods[tools.POST]; !ok {
return errors.New("no path found")
}
res, code, _ := a.LoadOne(id)
if code != 200 {
return errors.New("could not load workflow")
}
r := res.(*Workflow)
g := r.Graph
if realData.Graph != nil { // if the graph is set, set it to the real data
g = realData.Graph
}
if g != nil && g.Links != nil && len(g.Links) > 0 { // if the graph is set and has links then book the workflow (even on ourselves)
isDCFound := []string{}
for _, link := range g.Links {
if ok, dc_id := realData.isDCLink(link); ok { // check if the link is a link between a compute and a resource booking is only on compute
if slices.Contains(isDCFound, dc_id) {
continue
} // if the compute is already found, skip it
isDCFound = append(isDCFound, dc_id)
dc, code, _ := a.computeResourceAccessor.LoadOne(dc_id)
if code != 200 {
continue
}
// CHECK BOOKING
peerID := dc.(*resources.ComputeResource).PeerID
if peerID == "" { // no peer id no booking
continue
}
// BOOKING ON PEER
_, err := (&peer.Peer{}).LaunchPeerExecution(peerID, "", tools.BOOKING, tools.POST,
(&workflow_execution.WorkflowExecutions{ // it's the standard model for booking see OC-PEER
WorkflowID: id, // set the workflow id "WHO"
ResourceID: dc_id, // set the compute id "WHERE"
Executions: execs, // set the executions to book "WHAT"
}).Serialize(), a.Caller)
if err != nil {
fmt.Println("BOOKING", err)
return err
}
}
}
}
return nil
}
/*
* share is a function that shares a workflow to the peers if the workflow is shared
*/
@@ -197,10 +69,12 @@ func (a *workflowMongoAccessor) share(realData *Workflow, delete bool, caller *t
if ok, _ := paccess.IsMySelf(); ok { // if the peer is the current peer, never share because it will create a loop
continue
}
if delete { // if the workflow is deleted, share the deletion
if delete { // if the workflow is deleted, share the deletion orderResourceAccessor utils.Accessor
history := NewHistory()
history.StoreOne(history.MapFromWorkflow(res.(*Workflow)))
_, err = paccess.LaunchPeerExecution(p, res.GetID(), tools.WORKFLOW, tools.DELETE, map[string]interface{}{}, caller)
_, err = paccess.LaunchPeerExecution(p, res.GetID(), tools.WORKFLOW, tools.DELETE,
map[string]interface{}{}, caller)
} else { // if the workflow is updated, share the update
_, err = paccess.LaunchPeerExecution(p, res.GetID(), tools.WORKFLOW, tools.PUT,
res.Serialize(res), caller)
@@ -212,99 +86,30 @@ func (a *workflowMongoAccessor) share(realData *Workflow, delete bool, caller *t
}
}
/*
* execution is a create or delete function for the workflow executions depending on the schedule of the workflow
*/
func (a *workflowMongoAccessor) execution(id string, realData *Workflow, delete bool) (int, error) {
nats := tools.NewNATSCaller() // create a new nats caller because executions are sent to the nats for daemons
mongo.MONGOService.DeleteMultiple(map[string]interface{}{
"state": 1, // only delete the scheduled executions only scheduled if executions are in progress or ended, they should not be deleted for registration
"workflow_id": id,
}, tools.WORKFLOW_EXECUTION.String())
err := a.book(id, realData, []*workflow_execution.WorkflowExecution{}) // delete the booking of the workflow on the peers
fmt.Println("DELETE BOOKING", err)
nats.SetNATSPub(tools.WORKFLOW.String(), tools.REMOVE, realData) // send the deletion to the nats
if err != nil {
return 409, err
}
execs, err := a.getExecutions(id, realData) // get the executions of the workflow
if err != nil {
return 422, err
}
if !realData.ScheduleActive || delete { // if the schedule is not active, delete the executions
execs = []*workflow_execution.WorkflowExecution{}
}
err = a.book(id, realData, execs) // book the workflow on the peers
fmt.Println("BOOKING", err)
if err != nil {
return 409, err // if the booking fails, return an error for integrity between peers
}
fmt.Println("BOOKING", delete)
for _, obj := range execs {
_, code, err := a.executionAccessor.StoreOne(obj)
fmt.Println("EXEC", code, err)
if code != 200 {
return code, err
}
}
nats.SetNATSPub(tools.WORKFLOW.String(), tools.CREATE, realData) // send the creation to the nats
return 200, nil
}
// UpdateOne updates a workflow in the database
func (a *workflowMongoAccessor) UpdateOne(set utils.DBObject, id string) (utils.DBObject, int, error) {
res, code, err := a.LoadOne(id)
if code != 200 {
return nil, 409, err
}
// avoid the update if the schedule is the same
avoid := set.(*Workflow).Schedule == nil || (res.(*Workflow).Schedule != nil && res.(*Workflow).ScheduleActive == set.(*Workflow).ScheduleActive && res.(*Workflow).Schedule.Start == set.(*Workflow).Schedule.Start && res.(*Workflow).Schedule.End == set.(*Workflow).Schedule.End && res.(*Workflow).Schedule.Cron == set.(*Workflow).Schedule.Cron)
res, code, err = utils.GenericUpdateOne(set, id, a, &Workflow{})
res, code, err := utils.GenericUpdateOne(set, id, a, &Workflow{})
if code != 200 {
return nil, code, err
}
workflow := res.(*Workflow)
if !avoid { // if the schedule is not avoided, update the executions
if code, err := a.execution(id, workflow, false); code != 200 {
return nil, code, errors.New("could not update the executions : " + err.Error())
}
}
fmt.Println("UPDATE", workflow.ScheduleActive, workflow.Schedule)
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
utils.GenericRawUpdateOne(workflow, id, a)
} // if the start date is passed, update the executions
}
a.execute(workflow, false, false) // update the workspace for the workflow
a.share(workflow, false, a.Caller) // share the update to the peers
a.execute(workflow, false, false) // update the workspace for the workflow
a.share(workflow, false, a.GetCaller()) // share the update to the peers
return res, code, nil
}
// StoreOne stores a workflow in the database
func (a *workflowMongoAccessor) StoreOne(data utils.DBObject) (utils.DBObject, int, error) {
d := data.(*Workflow)
if d.ScheduleActive && d.Schedule != nil { // if the workflow is scheduled, update the executions
now := time.Now().UTC()
if (d.Schedule.End != nil && now.After(*d.Schedule.End)) || (d.Schedule.End == nil && d.Schedule.Start != nil && now.After(*d.Schedule.Start)) { // if the start date is passed, then you can book
d.ScheduleActive = false
} // if the start date is passed, update the executions
}
res, code, err := utils.GenericStoreOne(d, a)
if err != nil || code != 200 {
return nil, code, err
}
workflow := res.(*Workflow)
a.share(workflow, false, a.Caller) // share the creation to the peers
//store the executions
if code, err := a.execution(res.GetID(), workflow, false); err != nil {
return nil, code, err
}
a.execute(workflow, false, false) // store the workspace for the workflow
a.share(workflow, false, a.GetCaller()) // share the creation to the peers
a.execute(workflow, false, false) // store the workspace for the workflow
return res, code, nil
}
@@ -322,7 +127,7 @@ func (a *workflowMongoAccessor) execute(workflow *Workflow, delete bool, active
"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: workflow.Name + "_workspace"}},
},
}
resource, _, err := a.workspaceAccessor.Search(filters, "")
resource, _, err := a.workspaceAccessor.Search(filters, "", workflow.IsDraft)
if delete { // if delete is set to true, delete the workspace
for _, r := range resource {
a.workspaceAccessor.DeleteOne(r.GetID())
@@ -358,22 +163,15 @@ func (a *workflowMongoAccessor) execute(workflow *Workflow, delete bool, active
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
utils.GenericRawUpdateOne(d, id, a)
} // if the start date is passed, update the executions
}
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) LoadAll(isDraft bool) ([]utils.ShallowDBObject, int, error) {
return utils.GenericLoadAll[*Workflow](func(d utils.DBObject) utils.ShallowDBObject { return &d.(*Workflow).AbstractObject }, isDraft, 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)
func (a *workflowMongoAccessor) Search(filters *dbs.Filters, search string, isDraft bool) ([]utils.ShallowDBObject, int, error) {
return utils.GenericSearch[*Workflow](filters, search, (&Workflow{}).GetObjectFilters(search), func(d utils.DBObject) utils.ShallowDBObject { return d }, isDraft, a)
}

23
models/workflow/workflow_schedule.go
View File

@@ -1,23 +0,0 @@
package workflow
import "time"
// WorkflowSchedule is a struct that contains the scheduling information of a workflow
type ScheduleMode int
const (
TASK ScheduleMode = iota
SERVICE
)
/*
* WorkflowSchedule is a struct that contains the scheduling information of a workflow
* It contains the mode of the schedule (Task or Service), the name of the schedule, the start and end time of the schedule and the cron expression
*/
type WorkflowSchedule struct {
Mode int64 `json:"mode" bson:"mode" validate:"required"` // Mode is the mode of the schedule (Task or Service)
Name string `json:"name" bson:"name" validate:"required"` // Name is the name of the schedule
Start *time.Time `json:"start" bson:"start" validate:"required,ltfield=End"` // Start is the start time of the schedule, is required and must be less than the End time
End *time.Time `json:"end,omitempty" bson:"end,omitempty"` // End is the end time of the schedule
Cron string `json:"cron,omitempty" bson:"cron,omitempty"` // here the cron format : ss mm hh dd MM dw task
}

4
models/workflow/workflow_test.go
View File

@@ -13,7 +13,7 @@ func TestStoreOneWorkflow(t *testing.T) {
AbstractObject: utils.AbstractObject{Name: "testWorkflow"},
}
wma := New(tools.WORKFLOW, "", "", nil, nil)
wma := NewAccessor(tools.APIRequest{})
id, _, _ := wma.StoreOne(&w)
assert.NotEmpty(t, id)
@@ -24,7 +24,7 @@ func TestLoadOneWorkflow(t *testing.T) {
AbstractObject: utils.AbstractObject{Name: "testWorkflow"},
}
wma := New(tools.WORKFLOW, "", "", nil, nil)
wma := NewAccessor(tools.APIRequest{})
new_w, _, _ := wma.StoreOne(&w)
assert.Equal(t, w, new_w)
}