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

Compare commits

merge into: core:bugfix/lag-booking
Branches Tags
core:main core:bugfix/booking-too-long core:bugfix/lag-booking core:bugfix/schedule-cron core:test-wf core:feature/admiralty core:debug/check-peer core:namespace core:payment core:issue#4 core:chart core:logs
...
pull from: core:bugfix/booking-too-long
Branches Tags
core:main core:bugfix/booking-too-long core:bugfix/lag-booking core:bugfix/schedule-cron core:test-wf core:feature/admiralty core:debug/check-peer core:namespace core:payment core:issue#4 core:chart core:logs

7 Commits
bugfix/lag ... bugfix/boo

Author SHA1 Message Date
pb
778ffa05a1 what is collaborative 2025-06-02 18:05:08 +02:00
pb
3c15907427 added id for logger 2025-06-02 10:34:58 +02:00
pb
9ae5f3b91d timing status checks 2025-05-28 18:19:07 +02:00
pb
3a2141aab5 adding log to measure search time 2025-05-28 16:22:26 +02:00
pb
6ab6383144 corrected an use of the original http caller instead of the deep copy 2025-05-27 18:07:00 +02:00
pb
690d60f9d6 corrected the getBooking function parameters 2025-05-27 16:13:46 +02:00
pb
da0de80afd Booking check and booking post have been transformed in goroutine to improve performance when booking several execution with cron expressions 2025-05-27 15:38:24 +02:00
3 changed files with 18 additions and 17 deletions
Expand all files Collapse all files

6
models/booking/booking.go
View File

@@ -4,6 +4,7 @@ import (
"time"
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/logs"
"cloud.o-forge.io/core/oc-lib/models/common/enum"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
@@ -39,6 +40,8 @@ func (wfa *Booking) Check(id string, start time.Time, end *time.Time, parrallelA
end = &e
}
accessor := NewAccessor(nil)
l := logs.GetLogger().With().Str("Search Check", "Booking").Logger()
l.Debug().Msg("Starting to search")
res, code, err := accessor.Search(&dbs.Filters{
And: map[string][]dbs.Filter{ // check if there is a booking on the same compute resource by filtering on the compute_resource_id, the state and the execution date
"resource_id": {{Operator: dbs.EQUAL.String(), Value: id}},
@@ -49,6 +52,9 @@ func (wfa *Booking) Check(id string, start time.Time, end *time.Time, parrallelA
},
},
}, "", wfa.IsDraft)
l.Debug().Msg("Search finished")
if code != 200 {
return false, err
}

23
models/workflow_execution/workflow_scheduler.go
View File

@@ -7,7 +7,6 @@ import (
"sync"
"time"
"cloud.o-forge.io/core/oc-lib/logs"
"cloud.o-forge.io/core/oc-lib/models/booking"
"cloud.o-forge.io/core/oc-lib/models/common/enum"
"cloud.o-forge.io/core/oc-lib/models/peer"
@@ -16,7 +15,6 @@ import (
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
"github.com/robfig/cron"
"github.com/rs/zerolog"
)
/*
@@ -55,8 +53,6 @@ func NewScheduler(start string, end string, durationInS float64, cron string) *W
}
func (ws *WorkflowSchedule) CheckBooking(wfID string, request *tools.APIRequest) (bool, *workflow.Workflow, []*WorkflowExecution, []*booking.Booking, error) {
l := logs.GetLogger().With().Str("SchedulerID", ws.UUID).Logger()
l.Debug().Msg("Checking booking")
if request.Caller == nil && request.Caller.URLS == nil && request.Caller.URLS[tools.BOOKING] == nil || request.Caller.URLS[tools.BOOKING][tools.GET] == "" {
return false, nil, []*WorkflowExecution{}, []*booking.Booking{}, errors.New("no caller defined")
}
@@ -75,14 +71,12 @@ func (ws *WorkflowSchedule) CheckBooking(wfID string, request *tools.APIRequest)
if ws.End != nil && ws.Start.Add(time.Duration(longest)*time.Second).After(*ws.End) {
ws.Warning = "The workflow may be too long to be executed in the given time frame, we will try to book it anyway\n"
}
l.Debug().Msg("Getting executions")
execs, err := ws.getExecutions(wf)
if err != nil {
return false, wf, []*WorkflowExecution{}, []*booking.Booking{}, err
}
bookings := []*booking.Booking{}
for i, exec := range execs {
l.Debug().Msg("looping throughs execs : " + string(i))
for _, exec := range execs {
bookings = append(bookings, exec.Book(ws.UUID, wfID, priceds)...)
}
@@ -90,7 +84,7 @@ func (ws *WorkflowSchedule) CheckBooking(wfID string, request *tools.APIRequest)
var m sync.Mutex
for _, b := range bookings {
go getBooking(l, b, request, wf, execs, bookings, errCh, &m)
go getBooking(b, request, wf, execs, bookings, errCh, &m)
}
for i := 0; i < len(bookings); i++ {
@@ -102,7 +96,7 @@ func (ws *WorkflowSchedule) CheckBooking(wfID string, request *tools.APIRequest)
return true, wf, execs, bookings, nil
}
func getBooking(l zerolog.Logger, b *booking.Booking, request *tools.APIRequest, wf *workflow.Workflow, execs []*WorkflowExecution, bookings []*booking.Booking, errCh chan error, m *sync.Mutex) {
func getBooking( b *booking.Booking, request *tools.APIRequest, wf *workflow.Workflow, execs []*WorkflowExecution, bookings []*booking.Booking, errCh chan error, m *sync.Mutex) {
m.Lock()
c, err := getCallerCopy(request, errCh)
@@ -112,15 +106,13 @@ func getBooking(l zerolog.Logger, b *booking.Booking, request *tools.APIRequest,
}
m.Unlock()
bl := l.With().Str("booking", b.UUID).Logger()
meth := request.Caller.URLS[tools.BOOKING][tools.GET]
meth := c.URLS[tools.BOOKING][tools.GET]
meth = strings.ReplaceAll(meth, ":id", b.ResourceID)
meth = strings.ReplaceAll(meth, ":start_date", b.ExpectedStartDate.Format("2006-01-02T15:04:05"))
meth = strings.ReplaceAll(meth, ":end_date", b.ExpectedEndDate.Format("2006-01-02T15:04:05"))
request.Caller.URLS[tools.BOOKING][tools.GET] = meth
bl.Debug().Msg("Get booking " + b.UUID + " on " + b.DestPeerID)
c.URLS[tools.BOOKING][tools.GET] = meth
_, err = (&peer.Peer{}).LaunchPeerExecution(b.DestPeerID, b.ResourceID, tools.BOOKING, tools.GET, nil, &c)
bl.Debug().Msg("Received response from Get booking " + b.UUID + " on " + b.DestPeerID)
if err != nil {
errCh <- err
return
@@ -195,11 +187,8 @@ func (ws *WorkflowSchedule) BookExecs(booking *booking.Booking, request *tools.A
}
m.Unlock()
l := logs.GetLogger().With().Str("SchedulerID", ws.UUID).Logger()
l.Debug().Msg("Booking " + booking.UUID + " on " + booking.DestPeerID)
_, err = (&peer.Peer{}).LaunchPeerExecution(booking.DestPeerID, "",
tools.BOOKING, tools.POST, booking.Serialize(booking), &c)
l.Debug().Msg("Received answer for booking " + booking.UUID + " on " + booking.DestPeerID)
if err != nil {
errCh <- err

6
tools/api.go
View File

@@ -7,7 +7,9 @@ import (
"cloud.o-forge.io/core/oc-lib/config"
"cloud.o-forge.io/core/oc-lib/dbs/mongo"
"cloud.o-forge.io/core/oc-lib/logs"
beego "github.com/beego/beego/v2/server/web"
"github.com/google/uuid"
)
type APIRequest struct {
@@ -134,6 +136,9 @@ func (a *API) CheckRemotePeer(url string) (State, map[string]int) {
// CheckRemoteAPIs checks the state of remote APIs from your proper OC
func (a *API) CheckRemoteAPIs(apis []DataType) (State, map[string]string, error) {
id := uuid.New()
l := logs.GetLogger().With().Str("id",id.String()).Logger()
l.Debug().Msg("Start checking")
// Check if the database is up
new := map[string]string{}
caller := NewHTTPCaller(map[DataType]map[METHOD]string{}) // Create a new http caller
@@ -142,6 +147,7 @@ func (a *API) CheckRemoteAPIs(apis []DataType) (State, map[string]string, error)
state := ALIVE
reachable := false
for _, api := range apis { // Check the state of each remote API in the list
l.Debug().Msg("Checking : " + api.String() + " at " + api.API())
var resp APIStatusResponse
b, err := caller.CallGet("http://"+api.API()+":8080", "/oc/version/status") // Call the status endpoint of the remote API (standard OC status endpoint)
if err != nil {