25 Commits

Author SHA1 Message Date
pb
41ddf1f7cb changed how url is consructed in LaunchPeerExecution by placing meth after peer url + dt 2025-03-05 14:05:04 +01:00
pb
18c723ee42 removed caller from checkPeerStatus() parameters by adding the path to url 2025-03-05 12:30:38 +01:00
pb
d4f000466f add print for url 2025-03-05 12:18:07 +01:00
pb
dbdccdb920 Corrected urlFormat() from peer_cache which constructed url with API's name at the end 2025-03-05 11:01:46 +01:00
pb
fd3fef72d3 correct DefaultAPI list 2025-03-03 10:55:00 +01:00
pb
1890fd4f71 added the resources used for admiralty in datacenter API 2025-03-03 10:33:37 +01:00
pb
95af3cb515 removed caller from checkPeerStatus() parameters by adding the path to url 2025-03-03 10:32:52 +01:00
pb
3acebc451e Added the required tag to ExecutionsID 2025-02-26 11:00:37 +01:00
mr
5111c9c8be discovery clear view 2025-02-19 15:29:42 +01:00
mr
3ecb0e9d96 set up auth for workspace 2025-02-19 11:41:52 +01:00
mr
b4a1766677 better load & peear cache for traefik 2025-02-19 11:03:12 +01:00
mr
241c6a5a08 casual debug, time added before change of state bookin & exec 2025-02-19 08:55:11 +01:00
mr
7c30633bde verifyAuthAction for instance 2025-02-18 14:02:29 +01:00
mr
81d3406305 verifyAuthAction for instance 2025-02-18 12:55:49 +01:00
mr
04f7537066 save 2025-02-18 12:39:16 +01:00
mr
6bf058ab5c save 2025-02-18 11:11:40 +01:00
mr
b771b5d25e save 2025-02-18 10:25:08 +01:00
mr
6e6ed4ea2c debug 2025-02-18 09:53:55 +01:00
mr
a098f0a672 conf 2025-02-18 09:01:21 +01:00
plm
5255ffc2f7 Merging issue#4 2025-01-10 17:43:31 +01:00
plm
fd1c579ec4 Removing required field on PeerId, see #7 2025-01-10 17:39:58 +01:00
plm
0f4adeea86 Same prefix for all builtin microservices in opencloud 2025-01-08 16:55:42 +01:00
plm
245f3adea3 Merge branch 'issue#4' 2024-12-16 09:18:58 +01:00
plm
21d08204b5 Fixing env based layer; not using onion builtin mechanism to preserve opencloud conf key/value format 2024-12-16 09:17:54 +01:00
plm
1de4888599 Remove extra underscore character; it breaks the env var loading 2024-12-10 14:01:47 +01:00
13 changed files with 86 additions and 42 deletions

View File

@@ -26,12 +26,12 @@ import (
func GetConfLoader() *onion.Onion {
logger := zerolog.New(os.Stdout).With().Timestamp().Logger()
AppName := GetAppName()
EnvPrefix := strings.ToUpper(AppName[0:2]+AppName[3:]) + "_"
EnvPrefix := "OC_"
defaultConfigFile := "/etc/oc/" + AppName[3:] + ".json"
localConfigFile := "./" + AppName[3:] + ".json"
var configFile string
var o *onion.Onion
l3 := onion.NewEnvLayerPrefix("_", EnvPrefix)
l3 := GetEnvVarLayer(EnvPrefix)
l2, err := onion.NewFileLayer(localConfigFile, nil)
if err == nil {
logger.Info().Msg("Local config file found " + localConfigFile + ", overriding default file")
@@ -54,3 +54,17 @@ func GetConfLoader() *onion.Onion {
}
return o
}
func GetEnvVarLayer(prefix string) onion.Layer {
envVars := make(map[string]interface{})
for _, e := range os.Environ() {
pair := strings.SplitN(e, "=", 2)
key := pair[0]
if strings.HasPrefix(key, prefix) {
envVars[strings.TrimPrefix(key, prefix)] = pair[1]
}
}
return onion.NewMapLayer(envVars)
}

View File

@@ -15,7 +15,7 @@ import (
*/
type Booking struct {
utils.AbstractObject // AbstractObject contains the basic fields of an object (id, name)
ExecutionsID string `json:"executions_id,omitempty" bson:"executions_id,omitempty"` // ExecutionsID is the ID of the executions
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
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"`

View File

@@ -51,13 +51,15 @@ func (a *bookingMongoAccessor) CopyOne(data utils.DBObject) (utils.DBObject, int
func (a *bookingMongoAccessor) LoadOne(id string) (utils.DBObject, int, error) {
return utils.GenericLoadOne[*Booking](id, func(d utils.DBObject) (utils.DBObject, int, error) {
if d.(*Booking).State == enum.DRAFT && time.Now().UTC().After(d.(*Booking).ExpectedStartDate) {
now := time.Now()
now = now.Add(time.Second * -60)
if d.(*Booking).State == enum.DRAFT && now.UTC().After(d.(*Booking).ExpectedStartDate) {
return utils.GenericDeleteOne(d.GetID(), a)
}
if (d.(*Booking).ExpectedEndDate) == nil {
d.(*Booking).State = enum.FORGOTTEN
utils.GenericRawUpdateOne(d, id, a)
} else if d.(*Booking).State == enum.SCHEDULED && time.Now().UTC().After(*&d.(*Booking).ExpectedStartDate) {
} else if d.(*Booking).State == enum.SCHEDULED && now.UTC().After(d.(*Booking).ExpectedStartDate) {
d.(*Booking).State = enum.DELAYED
utils.GenericRawUpdateOne(d, id, a)
}
@@ -75,11 +77,13 @@ func (a *bookingMongoAccessor) Search(filters *dbs.Filters, search string, isDra
func (a *bookingMongoAccessor) getExec() func(utils.DBObject) utils.ShallowDBObject {
return func(d utils.DBObject) utils.ShallowDBObject {
if d.(*Booking).State == enum.DRAFT && time.Now().UTC().After(d.(*Booking).ExpectedStartDate) {
now := time.Now()
now = now.Add(time.Second * -60)
if d.(*Booking).State == enum.DRAFT && now.UTC().After(d.(*Booking).ExpectedStartDate) {
utils.GenericDeleteOne(d.GetID(), a)
return nil
}
if d.(*Booking).State == enum.SCHEDULED && time.Now().UTC().After(*&d.(*Booking).ExpectedStartDate) {
if d.(*Booking).State == enum.SCHEDULED && now.UTC().After(d.(*Booking).ExpectedStartDate) {
d.(*Booking).State = enum.DELAYED
utils.GenericRawUpdateOne(d, d.GetID(), a)
}

View File

@@ -29,47 +29,40 @@ type PeerCache struct {
}
// urlFormat formats the URL of the peer with the data type API function
func (p *PeerCache) urlFormat(url string, dt tools.DataType) string {
func (p *PeerCache) urlFormat(hostUrl string, dt tools.DataType) string {
// localhost is replaced by the local peer URL
// because localhost must collide on a web request security protocol
localhost := ""
if strings.Contains(url, "localhost") {
if strings.Contains(hostUrl, "localhost") {
localhost = "localhost"
}
if strings.Contains(url, "127.0.0.1") {
if strings.Contains(hostUrl, "127.0.0.1") {
localhost = "127.0.0.1"
}
if localhost != "" {
r := regexp.MustCompile("(" + localhost + ":[0-9]+)")
t := r.FindString(url)
t := r.FindString(hostUrl)
if t != "" {
url = strings.Replace(url, t, dt.API()+":8080/oc", -1)
hostUrl = strings.Replace(hostUrl, t, dt.API()+":8080/oc", -1)
} else {
url = strings.ReplaceAll(url, localhost, dt.API()+":8080/oc")
hostUrl = strings.ReplaceAll(hostUrl, localhost, dt.API()+":8080/oc")
}
} else {
url = url + "/" + dt.API()
hostUrl = hostUrl + "/" + strings.ReplaceAll(dt.API(), "oc-", "")
}
return url
fmt.Println("Host URL : ", hostUrl)
return hostUrl
}
// checkPeerStatus checks the status of a peer
func (p *PeerCache) checkPeerStatus(peerID string, appName string, caller *tools.HTTPCaller) (*Peer, bool) {
func (p *PeerCache) checkPeerStatus(peerID string, appName string) (*Peer, bool) {
api := tools.API{}
access := NewShallowAccessor()
res, code, _ := access.LoadOne(peerID) // Load the peer from db
if code != 200 { // no peer no party
return nil, false
}
methods := caller.URLS[tools.PEER] // Get the methods url of the peer
if methods == nil {
return res.(*Peer), false
}
meth := methods[tools.POST] // Get the POST method to check status
if meth == "" {
return res.(*Peer), false
}
url := p.urlFormat(res.(*Peer).Url, tools.PEER) + meth // Format the URL
url := p.urlFormat(res.(*Peer).Url, tools.PEER) + "/status" // Format the URL
state, services := api.CheckRemotePeer(url)
res.(*Peer).ServicesState = services // Update the services states of the peer
access.UpdateOne(res, peerID) // Update the peer in the db
@@ -89,11 +82,11 @@ func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string,
url := ""
// Check the status of the peer
if mypeer, ok := p.checkPeerStatus(peerID, dt.API(), caller); !ok && mypeer != nil {
if mypeer, ok := p.checkPeerStatus(peerID, dt.API()); !ok && mypeer != nil {
// If the peer is not reachable, add the execution to the failed executions list
pexec := &PeerExecution{
Method: method.String(),
Url: p.urlFormat((mypeer.Url)+meth, dt),
Url: p.urlFormat((mypeer.Url), dt) + meth,
Body: body,
DataType: dt.EnumIndex(),
DataID: dataID,
@@ -106,7 +99,7 @@ func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string,
return nil, errors.New("peer not found")
}
// If the peer is reachable, launch the execution
url = p.urlFormat((mypeer.Url)+meth, dt) // Format the URL
url = p.urlFormat((mypeer.Url), dt) + meth // Format the URL
tmp := mypeer.FailedExecution // Get the failed executions list
mypeer.FailedExecution = []PeerExecution{} // Reset the failed executions list
NewShallowAccessor().UpdateOne(mypeer, peerID) // Update the peer in the db

View File

@@ -89,6 +89,9 @@ func (r *AbstractInstanciatedResource[T]) GetSelectedInstance() utils.DBObject {
}
func (abs *AbstractInstanciatedResource[T]) SetAllowedInstances(request *tools.APIRequest) {
if request != nil && request.PeerID == abs.CreatorID && request.PeerID != "" {
return
}
abs.Instances = verifyAuthAction[T](abs.Instances, request)
}

View File

@@ -65,6 +65,12 @@ func (wfa *resourceMongoAccessor[T]) LoadAll(isDraft bool) ([]utils.ShallowDBObj
}
func (wfa *resourceMongoAccessor[T]) Search(filters *dbs.Filters, search string, isDraft bool) ([]utils.ShallowDBObject, int, error) {
if filters == nil && search == "*" {
return utils.GenericLoadAll[T](func(d utils.DBObject) utils.ShallowDBObject {
d.(T).SetAllowedInstances(wfa.Request)
return d
}, isDraft, wfa)
}
return utils.GenericSearch[T](filters, search, wfa.getResourceFilter(search),
func(d utils.DBObject) utils.ShallowDBObject {
d.(T).SetAllowedInstances(wfa.Request)
@@ -73,9 +79,6 @@ func (wfa *resourceMongoAccessor[T]) Search(filters *dbs.Filters, search string,
}
func (abs *resourceMongoAccessor[T]) getResourceFilter(search string) *dbs.Filters {
if search == "*" {
search = ""
}
return &dbs.Filters{
Or: map[string][]dbs.Filter{ // filter by like name, short_description, description, owner, url if no filters are provided
"abstractintanciatedresource.abstractresource.abstractobject.name": {{Operator: dbs.LIKE.String(), Value: search}},

View File

@@ -91,7 +91,7 @@ func (ao *AbstractObject) UpToDate(user string, peer string, create bool) {
}
func (ao *AbstractObject) VerifyAuth(request *tools.APIRequest) bool {
return ao.AccessMode == Public || (request != nil && ao.CreatorID == request.PeerID)
return ao.AccessMode == Public || (request != nil && ao.CreatorID == request.PeerID && request.PeerID != "")
}
func (ao *AbstractObject) GetObjectFilters(search string) *dbs.Filters {

View File

@@ -95,7 +95,7 @@ func (a *workflowMongoAccessor) UpdateOne(set utils.DBObject, id string) (utils.
if set.(*Workflow).Graph != nil && set.(*Workflow).Graph.Partial {
return nil, 403, errors.New("you are not allowed to update a partial workflow")
}
res, code, err := utils.GenericUpdateOne(a.verifyResource(set), id, a, &Workflow{})
res, code, err := utils.GenericUpdateOne(set, id, a, &Workflow{})
if code != 200 {
return nil, code, err
}

View File

@@ -60,11 +60,13 @@ func (wfa *workflowExecutionMongoAccessor) CopyOne(data utils.DBObject) (utils.D
func (a *workflowExecutionMongoAccessor) LoadOne(id string) (utils.DBObject, int, error) {
return utils.GenericLoadOne[*WorkflowExecution](id, func(d utils.DBObject) (utils.DBObject, int, error) {
if d.(*WorkflowExecution).State == enum.DRAFT && !a.shallow && time.Now().UTC().After(d.(*WorkflowExecution).ExecDate) {
now := time.Now()
now = now.Add(time.Second * -60)
if d.(*WorkflowExecution).State == enum.DRAFT && !a.shallow && now.UTC().After(d.(*WorkflowExecution).ExecDate) {
utils.GenericDeleteOne(d.GetID(), newShallowAccessor(a.Request))
return nil, 404, errors.New("not found")
}
if d.(*WorkflowExecution).State == enum.SCHEDULED && !a.shallow && time.Now().UTC().After(d.(*WorkflowExecution).ExecDate) {
if d.(*WorkflowExecution).State == enum.SCHEDULED && !a.shallow && now.UTC().After(d.(*WorkflowExecution).ExecDate) {
d.(*WorkflowExecution).State = enum.FORGOTTEN
utils.GenericRawUpdateOne(d, id, newShallowAccessor(a.Request))
}
@@ -82,11 +84,13 @@ func (a *workflowExecutionMongoAccessor) Search(filters *dbs.Filters, search str
func (a *workflowExecutionMongoAccessor) getExec() func(utils.DBObject) utils.ShallowDBObject {
return func(d utils.DBObject) utils.ShallowDBObject {
if d.(*WorkflowExecution).State == enum.DRAFT && time.Now().UTC().After(d.(*WorkflowExecution).ExecDate) {
now := time.Now()
now = now.Add(time.Second * -60)
if d.(*WorkflowExecution).State == enum.DRAFT && now.UTC().After(d.(*WorkflowExecution).ExecDate) {
utils.GenericDeleteOne(d.GetID(), newShallowAccessor(a.Request))
return nil
}
if d.(*WorkflowExecution).State == enum.SCHEDULED && time.Now().UTC().After(d.(*WorkflowExecution).ExecDate) {
if d.(*WorkflowExecution).State == enum.SCHEDULED && now.UTC().After(d.(*WorkflowExecution).ExecDate) {
d.(*WorkflowExecution).State = enum.FORGOTTEN
utils.GenericRawUpdateOne(d, d.GetID(), newShallowAccessor(a.Request))
return d

View File

@@ -118,15 +118,16 @@ func (ws *WorkflowSchedule) Schedules(wfID string, request *tools.APIRequest) (*
return ws, wf, executions, errors.New("could not launch the peer execution : " + fmt.Sprintf("%v", err))
}
}
fmt.Println("Schedules")
for _, exec := range executions {
err := exec.PurgeDraft(request)
if err != nil {
return ws, nil, []*WorkflowExecution{}, errors.New("purge draft" + fmt.Sprintf("%v", err))
}
exec.StoreDraftDefault()
// Should DELETE the previous execution2
fmt.Println(utils.GenericStoreOne(exec, NewAccessor(request)))
utils.GenericStoreOne(exec, NewAccessor(request))
}
fmt.Println("Schedules")
return ws, wf, executions, nil
}

View File

@@ -73,8 +73,10 @@ func (a *workspaceMongoAccessor) StoreOne(data utils.DBObject) (utils.DBObject,
filters := &dbs.Filters{
Or: map[string][]dbs.Filter{
"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: data.GetName() + "_workspace"}},
"abstractobject.creator_id": {{Operator: dbs.EQUAL.String(), Value: a.GetPeerID()}},
},
}
// filters *dbs.Filters, word string, isDraft bool
res, _, err := a.Search(filters, "", true) // Search for the workspace
if err == nil && len(res) > 0 { // If the workspace already exists, return an error
return nil, 409, errors.New("a workspace with the same name already exists")

View File

@@ -115,8 +115,8 @@ func (a *API) SubscribeRouter(infos []*beego.ControllerInfo) {
// CheckRemotePeer checks the state of a remote peer
func (a *API) CheckRemotePeer(url string) (State, map[string]int) {
// Check if the database is up
caller := NewHTTPCaller(map[DataType]map[METHOD]string{}) // Create a new http caller
var resp APIStatusResponse
caller := NewHTTPCaller(map[DataType]map[METHOD]string{}) // Create a new http caller
b, err := caller.CallPost(url, "", map[string]interface{}{}) // Call the status endpoint of the peer
if err != nil {
return DEAD, map[string]int{} // If the peer is not reachable, return dead

View File

@@ -21,6 +21,11 @@ const (
WORKSPACE_HISTORY
ORDER
PURCHASE_RESOURCE
ADMIRALTY_SOURCE
ADMIRALTY_TARGET
ADMIRALTY_SECRET
ADMIRALTY_KUBECONFIG
ADMIRALTY_NODES
)
var NOAPI = ""
@@ -30,6 +35,11 @@ var WORKFLOWAPI = "oc-workflow"
var WORKSPACEAPI = "oc-workspace"
var PEERSAPI = "oc-peer"
var DATACENTERAPI = "oc-datacenter"
var ADMIRALTY_SOURCEAPI = DATACENTERAPI+"/admiralty/source"
var ADMIRALTY_TARGETAPI = DATACENTERAPI+"/admiralty/target"
var ADMIRALTY_SECRETAPI = DATACENTERAPI+"/admiralty/secret"
var ADMIRALTY_KUBECONFIGAPI = DATACENTERAPI+"/admiralty/kubeconfig"
var ADMIRALTY_NODESAPI = DATACENTERAPI+"/admiralty/nodes"
// Bind the standard API name to the data type
var DefaultAPI = [...]string{
@@ -50,6 +60,11 @@ var DefaultAPI = [...]string{
NOAPI,
NOAPI,
NOAPI,
ADMIRALTY_SOURCEAPI,
ADMIRALTY_TARGETAPI,
ADMIRALTY_SECRETAPI,
ADMIRALTY_KUBECONFIGAPI,
ADMIRALTY_NODESAPI,
}
// Bind the standard data name to the data type
@@ -71,6 +86,11 @@ var Str = [...]string{
"workspace_history",
"order",
"purchase_resource",
"admiralty_source",
"admiralty_target",
"admiralty_secret",
"admiralty_kubeconfig",
"admiralty_node",
}
func FromInt(i int) string {
@@ -91,5 +111,5 @@ func (d DataType) EnumIndex() int {
}
func DataTypeList() []DataType {
return []DataType{DATA_RESOURCE, PROCESSING_RESOURCE, STORAGE_RESOURCE, COMPUTE_RESOURCE, WORKFLOW_RESOURCE, WORKFLOW, WORKFLOW_EXECUTION, WORKSPACE, PEER, COLLABORATIVE_AREA, RULE, BOOKING, WORKFLOW_HISTORY, WORKSPACE_HISTORY, ORDER, PURCHASE_RESOURCE}
return []DataType{DATA_RESOURCE, PROCESSING_RESOURCE, STORAGE_RESOURCE, COMPUTE_RESOURCE, WORKFLOW_RESOURCE, WORKFLOW, WORKFLOW_EXECUTION, WORKSPACE, PEER, COLLABORATIVE_AREA, RULE, BOOKING, WORKFLOW_HISTORY, WORKSPACE_HISTORY, ORDER, PURCHASE_RESOURCE,ADMIRALTY_SOURCE,ADMIRALTY_TARGET,ADMIRALTY_SECRET,ADMIRALTY_KUBECONFIG,ADMIRALTY_NODES}
}