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

booking start intelligency

Browse Source
This commit is contained in:
mr
2024-08-12 14:18:13 +02:00
parent eea7f25379
commit 93b10de61d
6 changed files with 272 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
models/workflow/workflow.go
View File

@@ -4,9 +4,12 @@ import (
"encoding/json"
"slices"
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/models/resources"
"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_execution"
"go.mongodb.org/mongo-driver/bson/primitive"
)
type AbstractWorkflow struct {
@@ -16,11 +19,14 @@ type AbstractWorkflow struct {
Shared []string `json:"shared,omitempty" bson:"shared,omitempty"`
}
func (w *AbstractWorkflow) isDCLink(link graph.GraphLink) bool {
if slices.Contains(w.Datacenters, link.Destination.ID) || slices.Contains(w.Datacenters, link.Source.ID) {
return true
func (w *AbstractWorkflow) isDCLink(link graph.GraphLink) (bool, string) {
if slices.Contains(w.Datacenters, link.Source.ID) {
return true, link.Source.ID
}
return false
if slices.Contains(w.Datacenters, link.Destination.ID) {
return true, link.Destination.ID
}
return false, ""
}
type Workflow struct {
@@ -28,35 +34,34 @@ type Workflow struct {
AbstractWorkflow
}
func (d *Workflow) GetName() string {
return d.Name
func (wfa *Workflow) CheckBooking() (bool, error) {
// check if
if wfa.Schedule == nil || wfa.Schedule.Start == nil {
return false, nil
}
if wfa.Schedule.End == nil {
// if no end... then Book like a savage
return true, nil
}
e := *wfa.Schedule.End
accessor := wfa.GetAccessor()
res, code, err := accessor.Search(&dbs.Filters{
And: map[string][]dbs.Filter{
"workflowexecution.state": {{Operator: dbs.EQUAL.String(), Value: workflow_execution.SCHEDULED.EnumIndex()}},
"workflowexecution.execution_date": {
{Operator: dbs.LTE.String(), Value: primitive.NewDateTimeFromTime(e)},
{Operator: dbs.GTE.String(), Value: primitive.NewDateTimeFromTime(*wfa.Schedule.Start)},
},
},
}, "")
if code != 200 {
return false, err
}
return len(res) == 0, nil
}
func (d *Workflow) CheckBooking() bool {
return true
/*if d.Schedule != nil && d.Schedule.Start != nil {
sd := primitive.NewDateTimeFromTime(d.Schedule.Start.Add(time.Minute * -1))
var f dbs.Filters
if d.Schedule.End == nil {
ed := primitive.NewDateTimeFromTime(d.Schedule.Start.Add(time.Minute * 10))
f = dbs.Filters{
And: map[string][]dbs.Filter{
"execution_date": {{Operator: "gte", Value: sd}, {Operator: "lte", Value: ed}},
},
}
} else {
ed := primitive.NewDateTimeFromTime(d.Schedule.End.Add(time.Minute * 1))
f = dbs.Filters{
And: map[string][]dbs.Filter{
"execution_date": {{Operator: "gte", Value: sd}},
"end_date": {{Operator: "lte", Value: ed}},
},
}
}
res, _, _ := (&workflow_execution.WorkflowExecution{}).GetAccessor().Search(&f, "")
return len(res) == 0
}
return true*/
func (d *Workflow) GetName() string {
return d.Name
}
func (d *Workflow) GetAccessor() utils.Accessor {

70
models/workflow/workflow_mongo_accessor.go
View File

@@ -6,7 +6,9 @@ import (
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/dbs/mongo"
"cloud.o-forge.io/core/oc-lib/models/booking"
"cloud.o-forge.io/core/oc-lib/models/resources"
"cloud.o-forge.io/core/oc-lib/models/resources/datacenter"
"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"
@@ -78,6 +80,60 @@ func (wfa *workflowMongoAccessor) DeleteOne(id string) (utils.DBObject, int, err
return wfa.GenericDeleteOne(id, wfa)
}
func (wfa *workflowMongoAccessor) book(id string, realData *Workflow, execs []*workflow_execution.WorkflowExecution) []*booking.Booking {
books := []*booking.Booking{}
if realData.Schedule == nil {
return books
}
res, _, _ := wfa.LoadOne(id)
r := res.(*Workflow)
g := r.Graph
if realData.Graph != nil {
g = realData.Graph
}
if g != nil && g.Links != nil && len(g.Links) > 0 {
bookAccessor := (&booking.Booking{}).GetAccessor()
accessor := (&datacenter.DatacenterResource{}).GetAccessor()
for _, link := range g.Links {
if ok, dc_id := realData.isDCLink(link); ok {
_, code, _ := accessor.LoadOne(dc_id)
if code != 200 {
continue
}
// CHECK BOOKING
// dc.(*datacenter.DatacenterResource).SourceUrl should get source url... but it's not implemented
res, code, _ := bookAccessor.Search(&dbs.Filters{And: map[string][]dbs.Filter{
"peer_id": {{Operator: dbs.EQUAL.String(), Value: "my_peer"}}, // peer is always the same for the moment
"datacenter_resource_id": {{Operator: dbs.EQUAL.String(), Value: dc_id}},
}}, "")
if code != 200 {
continue
}
for _, b := range res {
bookAccessor.DeleteOne(b.GetID())
}
for _, exec := range execs {
if ok, err := (&booking.Booking{}).CheckBooking(*exec.ExecDate, exec.EndDate); !ok {
if err != nil {
return books
}
return books
}
b, code, _ := bookAccessor.StoreOne(&booking.Booking{
PeerID: "my_peer",
DatacenterResourceID: dc_id,
WorkflowExecution: *exec,
})
if code == 200 {
books = append(books, b.(*booking.Booking))
}
}
}
}
}
return books
}
func (wfa *workflowMongoAccessor) execution(id string, realData *Workflow, delete bool) (int, error) {
if realData.Schedule == nil {
return 200, nil
@@ -87,8 +143,15 @@ func (wfa *workflowMongoAccessor) execution(id string, realData *Workflow, delet
if r.Schedule != nil && r.Schedule.Start == realData.Schedule.Start && r.Schedule.End == realData.Schedule.End && r.Schedule.Cron == realData.Schedule.Cron {
return 200, nil
}
if !realData.CheckBooking() {
return 409, errors.New("the booking is already taken.")
accessor := (&workflow_execution.WorkflowExecution{}).GetAccessor()
execs, err := wfa.getExecutions(id, realData)
for _, exec := range execs {
if ok, err := (&booking.Booking{}).CheckBooking(*exec.ExecDate, exec.EndDate); !ok {
if err != nil {
return 500, err
}
return 409, errors.New("the booking from " + exec.ExecDate.String() + " is already taken.")
}
}
if delete {
mongo.MONGOService.DeleteMultiple(map[string]interface{}{
@@ -96,8 +159,6 @@ func (wfa *workflowMongoAccessor) execution(id string, realData *Workflow, delet
"state": 1,
}, utils.WORKFLOW_EXECUTION.String())
}
accessor := (&workflow_execution.WorkflowExecution{}).GetAccessor()
execs, err := wfa.getExecutions(id, realData)
if err == nil && len(execs) > 0 {
for _, obj := range execs {
_, code, err := accessor.StoreOne(obj)
@@ -108,6 +169,7 @@ func (wfa *workflowMongoAccessor) execution(id string, realData *Workflow, delet
} else {
return 422, err
}
wfa.book(id, realData, execs)
return 200, nil
}