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

implement remote call for remote action

Browse Source
This commit is contained in:
mr 2024-08-12 16:11:25 +02:00
parent 4575f9ad3f
commit 2ac24779cd
26 changed files with 312 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
dbs/dbs.go
View File

@ -18,6 +18,7 @@ const (
LT
GT
EQUAL
NOT
)
var str = [...]string{
@ -29,6 +30,7 @@ var str = [...]string{
"lt",
"gt",
"equal",
"not",
}
func (m Operator) String() string {
@ -36,6 +38,12 @@ func (m Operator) String() string {
}
func (m Operator) ToMongoEOperator(k string, value interface{}) bson.E {
defer func() {
if r := recover(); r != nil {
fmt.Println("Recovered. Error:\n", r)
}
}()
defaultValue := bson.E{Key: k, Value: bson.M{"$regex": ToValueOperator(StringToOperator(m.String()), value)}}
switch m {
case LIKE:
return bson.E{Key: k, Value: bson.M{"$regex": ToValueOperator(StringToOperator(m.String()), value)}}
@ -53,12 +61,43 @@ func (m Operator) ToMongoEOperator(k string, value interface{}) bson.E {
return bson.E{Key: k, Value: bson.M{"$lt": ToValueOperator(StringToOperator(m.String()), value)}}
case EQUAL:
return bson.E{Key: k, Value: value}
case NOT:
v := value.(Filters)
orList := bson.A{}
andList := bson.A{}
f := bson.D{}
for k, filter := range v.Or {
for _, ff := range filter {
orList = append(orList, StringToOperator(ff.Operator).ToMongoOperator(k, ff.Value))
}
}
for k, filter := range v.And {
for _, ff := range filter {
andList = append(andList, StringToOperator(ff.Operator).ToMongoOperator(k, ff.Value))
}
}
if len(orList) > 0 && len(andList) == 0 {
f = bson.D{{"$or", orList}}
} else {
if len(orList) > 0 {
andList = append(andList, bson.M{"$or": orList})
}
f = bson.D{{"$and", andList}}
}
return bson.E{Key: "$not", Value: f}
default:
return bson.E{Key: k, Value: bson.M{"$regex": ToValueOperator(StringToOperator(m.String()), value)}}
return defaultValue
}
return defaultValue
}
func (m Operator) ToMongoOperator(k string, value interface{}) bson.M {
defer func() {
if r := recover(); r != nil {
fmt.Println("Recovered. Error:\n", r)
}
}()
defaultValue := bson.M{k: bson.M{"$regex": ToValueOperator(StringToOperator(m.String()), value)}}
switch m {
case LIKE:
return bson.M{k: bson.M{"$regex": ToValueOperator(StringToOperator(m.String()), value)}}
@ -76,9 +115,34 @@ func (m Operator) ToMongoOperator(k string, value interface{}) bson.M {
return bson.M{k: bson.M{"$lt": ToValueOperator(StringToOperator(m.String()), value)}}
case EQUAL:
return bson.M{k: value}
case NOT:
v := value.(Filters)
orList := bson.A{}
andList := bson.A{}
f := bson.D{}
for k, filter := range v.Or {
for _, ff := range filter {
orList = append(orList, StringToOperator(ff.Operator).ToMongoOperator(k, ff.Value))
}
}
for k, filter := range v.And {
for _, ff := range filter {
andList = append(andList, StringToOperator(ff.Operator).ToMongoOperator(k, ff.Value))
}
}
if len(orList) > 0 && len(andList) == 0 {
f = bson.D{{"$or", orList}}
} else {
if len(orList) > 0 {
andList = append(andList, bson.M{"$or": orList})
}
f = bson.D{{"$and", andList}}
}
return bson.M{"$not": f}
default:
return bson.M{k: bson.M{"$regex": ToValueOperator(StringToOperator(m.String()), value)}}
return defaultValue
}
return defaultValue
}
func StringToOperator(s string) Operator {

38
entrypoint.go
View File

@ -65,7 +65,7 @@ func Init(appName string) {
logs.SetAppName(appName)
logs.SetLogger(logs.CreateLogger("main", ""))
mongo.MONGOService.Init(models.GetModelsNames(), GetConfig())
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
for _, model := range []string{utils.DATA_RESOURCE.String(), utils.PROCESSING_RESOURCE.String(), utils.STORAGE_RESOURCE.String(), utils.DATACENTER_RESOURCE.String(), utils.WORKFLOW_RESOURCE.String()} {
data, code, _ := accessor.Search(nil, model)
if code == 404 || len(data) == 0 {
@ -97,7 +97,7 @@ func GetLogger() zerolog.Logger {
}
func Search(filters *dbs.Filters, word string, collection LibDataEnum) LibDataShallow {
d, code, err := models.Model(collection.EnumIndex()).GetAccessor().Search(filters, word)
d, code, err := models.Model(collection.EnumIndex()).GetAccessor(nil).Search(filters, word)
if err != nil {
return LibDataShallow{Data: d, Code: code, Err: err.Error()}
}
@ -105,7 +105,7 @@ func Search(filters *dbs.Filters, word string, collection LibDataEnum) LibDataSh
}
func LoadAll(collection LibDataEnum) LibDataShallow {
d, code, err := models.Model(collection.EnumIndex()).GetAccessor().LoadAll()
d, code, err := models.Model(collection.EnumIndex()).GetAccessor(nil).LoadAll()
if err != nil {
return LibDataShallow{Data: d, Code: code, Err: err.Error()}
}
@ -113,7 +113,7 @@ func LoadAll(collection LibDataEnum) LibDataShallow {
}
func LoadOne(collection LibDataEnum, id string) LibData {
d, code, err := models.Model(collection.EnumIndex()).GetAccessor().LoadOne(id)
d, code, err := models.Model(collection.EnumIndex()).GetAccessor(nil).LoadOne(id)
if err != nil {
return LibData{Data: d, Code: code, Err: err.Error()}
}
@ -122,7 +122,7 @@ func LoadOne(collection LibDataEnum, id string) LibData {
func UpdateOne(collection LibDataEnum, set map[string]interface{}, id string) LibData {
model := models.Model(collection.EnumIndex())
d, code, err := model.GetAccessor().UpdateOne(model.Deserialize(set), id)
d, code, err := model.GetAccessor(nil).UpdateOne(model.Deserialize(set), id)
if err != nil {
return LibData{Data: d, Code: code, Err: err.Error()}
}
@ -130,7 +130,7 @@ func UpdateOne(collection LibDataEnum, set map[string]interface{}, id string) Li
}
func DeleteOne(collection LibDataEnum, id string) LibData {
d, code, err := models.Model(collection.EnumIndex()).GetAccessor().DeleteOne(id)
d, code, err := models.Model(collection.EnumIndex()).GetAccessor(nil).DeleteOne(id)
if err != nil {
return LibData{Data: d, Code: code, Err: err.Error()}
}
@ -139,7 +139,7 @@ func DeleteOne(collection LibDataEnum, id string) LibData {
func StoreOne(collection LibDataEnum, object map[string]interface{}) LibData {
model := models.Model(collection.EnumIndex())
d, code, err := model.GetAccessor().StoreOne(model.Deserialize(object))
d, code, err := model.GetAccessor(nil).StoreOne(model.Deserialize(object))
if err != nil {
return LibData{Data: d, Code: code, Err: err.Error()}
}
@ -148,7 +148,7 @@ func StoreOne(collection LibDataEnum, object map[string]interface{}) LibData {
func CopyOne(collection LibDataEnum, object map[string]interface{}) LibData {
model := models.Model(collection.EnumIndex())
d, code, err := model.GetAccessor().CopyOne(model.Deserialize(object))
d, code, err := model.GetAccessor(nil).CopyOne(model.Deserialize(object))
if err != nil {
return LibData{Data: d, Code: code, Err: err.Error()}
}
@ -158,72 +158,72 @@ func CopyOne(collection LibDataEnum, object map[string]interface{}) LibData {
// ================ CAST ========================= //
func (l *LibData) ToDataResource() *data.DataResource {
if l.Data.GetAccessor().GetType() == utils.DATA_RESOURCE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.DATA_RESOURCE.String() {
return l.Data.(*data.DataResource)
}
return nil
}
func (l *LibData) ToDatacenterResource() *datacenter.DatacenterResource {
if l.Data != nil && l.Data.GetAccessor().GetType() == utils.DATACENTER_RESOURCE.String() {
if l.Data != nil && l.Data.GetAccessor(nil).GetType() == utils.DATACENTER_RESOURCE.String() {
return l.Data.(*datacenter.DatacenterResource)
}
return nil
}
func (l *LibData) ToStorageResource() *storage.StorageResource {
if l.Data.GetAccessor().GetType() == utils.STORAGE_RESOURCE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.STORAGE_RESOURCE.String() {
return l.Data.(*storage.StorageResource)
}
return nil
}
func (l *LibData) ToProcessingResource() *processing.ProcessingResource {
if l.Data.GetAccessor().GetType() == utils.PROCESSING_RESOURCE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.PROCESSING_RESOURCE.String() {
return l.Data.(*processing.ProcessingResource)
}
return nil
}
func (l *LibData) ToWorkflowResource() *w.WorkflowResource {
if l.Data.GetAccessor().GetType() == utils.WORKFLOW_RESOURCE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.WORKFLOW_RESOURCE.String() {
return l.Data.(*w.WorkflowResource)
}
return nil
}
func (l *LibData) ToPeer() *peer.Peer {
if l.Data.GetAccessor().GetType() == utils.PEER.String() {
if l.Data.GetAccessor(nil).GetType() == utils.PEER.String() {
return l.Data.(*peer.Peer)
}
return nil
}
func (l *LibData) ToWorkflow() *w2.Workflow {
if l.Data.GetAccessor().GetType() == utils.WORKFLOW.String() {
if l.Data.GetAccessor(nil).GetType() == utils.WORKFLOW.String() {
return l.Data.(*w2.Workflow)
}
return nil
}
func (l *LibData) ToWorkspace() *workspace.Workspace {
if l.Data.GetAccessor().GetType() == utils.WORKSPACE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.WORKSPACE.String() {
return l.Data.(*workspace.Workspace)
}
return nil
}
func (l *LibData) ToSharedWorkspace() *shared_workspace.SharedWorkspace {
if l.Data.GetAccessor().GetType() == utils.SHARED_WORKSPACE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.SHARED_WORKSPACE.String() {
return l.Data.(*shared_workspace.SharedWorkspace)
}
return nil
}
func (l *LibData) ToRule() *rule.Rule {
if l.Data.GetAccessor().GetType() == utils.SHARED_WORKSPACE.String() {
if l.Data.GetAccessor(nil).GetType() == utils.SHARED_WORKSPACE.String() {
return l.Data.(*rule.Rule)
}
return nil
}
func (l *LibData) ToWorkflowExecution() *workflow_execution.WorkflowExecution {
if l.Data.GetAccessor().GetType() == utils.WORKFLOW_EXECUTION.String() {
if l.Data.GetAccessor(nil).GetType() == utils.WORKFLOW_EXECUTION.String() {
return l.Data.(*workflow_execution.WorkflowExecution)
}
return nil

8
models/booking/booking.go
View File

@ -7,13 +7,13 @@ import (
"cloud.o-forge.io/core/oc-lib/dbs"
"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/tools"
"github.com/google/uuid"
"go.mongodb.org/mongo-driver/bson/primitive"
)
type Booking struct {
workflow_execution.WorkflowExecution
PeerID string `json:"peer_id,omitempty" bson:"peer_id,omitempty" validate:"required"`
DatacenterResourceID string `json:"datacenter_resource_id,omitempty" bson:"datacenter_resource_id,omitempty" validate:"required"`
}
@ -24,7 +24,7 @@ func (wfa *Booking) CheckBooking(start time.Time, end *time.Time) (bool, error)
return true, nil
}
e := *end
accessor := wfa.GetAccessor()
accessor := wfa.GetAccessor(nil)
res, code, err := accessor.Search(&dbs.Filters{
And: map[string][]dbs.Filter{
"workflowexecution.state": {{Operator: dbs.EQUAL.String(), Value: workflow_execution.SCHEDULED.EnumIndex()}},
@ -57,9 +57,9 @@ func (d *Booking) GetName() string {
return d.UUID + "_" + d.ExecDate.String()
}
func (d *Booking) GetAccessor() utils.Accessor {
func (d *Booking) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.BOOKING)
data.Init(utils.BOOKING, caller)
return data
}

5
models/peer/peer.go
View File

@ -4,6 +4,7 @@ import (
"encoding/json"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
)
@ -25,9 +26,9 @@ func (d *Peer) GetName() string {
return d.Name
}
func (d *Peer) GetAccessor() utils.Accessor {
func (d *Peer) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.PEER)
data.Init(utils.PEER, caller)
return data
}

7
models/resource_model/resource_model.go
View File

@ -4,6 +4,7 @@ import (
"encoding/json"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
)
@ -15,7 +16,7 @@ type AbstractResource struct {
Owner string `json:"owner,omitempty" bson:"owner,omitempty" validate:"required"`
OwnerLogo string `json:"owner_logo,omitempty" bson:"owner_logo,omitempty"`
SourceUrl string `json:"source_url,omitempty" bson:"source_url,omitempty" validate:"required"`
PeerID string `json:"peer_id,omitempty" bson:"peer_id,omitempty" validate:"required"`
PeerID string `json:"peer_id,omitempty" bson:"peer_id,omitempty" validate:"required"`
Price string `json:"price,omitempty" bson:"price,omitempty"`
License string `json:"license,omitempty" bson:"license,omitempty"`
ResourceModel *ResourceModel `json:"resource_model,omitempty" bson:"resource_model,omitempty"`
@ -83,9 +84,9 @@ func (d *ResourceModel) GetName() string {
return d.UUID
}
func (d *ResourceModel) GetAccessor() utils.Accessor {
func (d *ResourceModel) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := &ResourceModelMongoAccessor{}
data.SetLogger(utils.RESOURCE_MODEL)
data.Init(utils.RESOURCE_MODEL, caller)
return data
}

5
models/resources/data/data.go
View File

@ -5,6 +5,7 @@ import (
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
)
type DataResource struct {
@ -33,8 +34,8 @@ func (dma *DataResource) Serialize() map[string]interface{} {
return m
}
func (d *DataResource) GetAccessor() utils.Accessor {
func (d *DataResource) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.DATA_RESOURCE)
data.Init(utils.DATA_RESOURCE, caller)
return data
}

6
models/resources/data/data_mongo_accessor.go
View File

@ -41,7 +41,7 @@ func (dma *dataMongoAccessor) LoadOne(id string) (utils.DBObject, int, error) {
return nil, code, err
}
res_mongo.Decode(&data)
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, dma.GetType())
if err == nil && len(resources) > 0 {
data.ResourceModel = resources[0].(*resource_model.ResourceModel)
@ -60,7 +60,7 @@ func (wfa dataMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int, error) {
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {
@ -93,7 +93,7 @@ func (wfa *dataMongoAccessor) Search(filters *dbs.Filters, search string) ([]uti
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {

5
models/resources/datacenter/datacenter.go
View File

@ -5,6 +5,7 @@ import (
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
)
type DatacenterResource struct {
@ -33,9 +34,9 @@ func (dma *DatacenterResource) Serialize() map[string]interface{} {
return m
}
func (d *DatacenterResource) GetAccessor() utils.Accessor {
func (d *DatacenterResource) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.DATACENTER_RESOURCE)
data.Init(utils.DATACENTER_RESOURCE, caller)
return data
}

6
models/resources/datacenter/datacenter_mongo_accessor.go
View File

@ -43,7 +43,7 @@ func (dca *datacenterMongoAccessor) LoadOne(id string) (utils.DBObject, int, err
}
res_mongo.Decode(&datacenter)
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, dca.GetType())
if err == nil && len(resources) > 0 {
datacenter.ResourceModel = resources[0].(*resource_model.ResourceModel)
@ -62,7 +62,7 @@ func (wfa datacenterMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int, erro
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {
@ -95,7 +95,7 @@ func (wfa *datacenterMongoAccessor) Search(filters *dbs.Filters, search string)
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {

7
models/resources/processing/processing.go
View File

@ -6,13 +6,14 @@ import (
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"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/tools"
)
type ProcessingResource struct {
resource_model.AbstractResource
CPUs []*datacenter.CPU `bson:"cpus,omitempty" json:"cp_us,omitempty"`
GPUs []*datacenter.GPU `bson:"gpus,omitempty" json:"gp_us,omitempty"`
RAM *datacenter.RAM `bson:"ram,omitempty" json:"ram,omitempty"`
RAM *datacenter.RAM `bson:"ram,omitempty" json:"ram,omitempty"`
Storage uint `bson:"storage,omitempty" json:"storage,omitempty"`
Parallel bool `bson:"parallel,omitempty" json:"parallel,omitempty"`
ScalingModel uint `bson:"scaling_model,omitempty" json:"scaling_model,omitempty"`
@ -38,8 +39,8 @@ func (dma *ProcessingResource) Serialize() map[string]interface{} {
return m
}
func (d *ProcessingResource) GetAccessor() utils.Accessor {
func (d *ProcessingResource) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.PROCESSING_RESOURCE)
data.Init(utils.PROCESSING_RESOURCE, caller)
return data
}

6
models/resources/processing/processing_mongo_accessor.go
View File

@ -44,7 +44,7 @@ func (pma *processingMongoAccessor) LoadOne(id string) (utils.DBObject, int, err
}
res_mongo.Decode(&processing)
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, pma.GetType())
if err == nil && len(resources) > 0 {
processing.ResourceModel = resources[0].(*resource_model.ResourceModel)
@ -63,7 +63,7 @@ func (wfa processingMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int, erro
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {
@ -96,7 +96,7 @@ func (wfa *processingMongoAccessor) Search(filters *dbs.Filters, search string)
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {

5
models/resources/storage/storage.go
View File

@ -5,6 +5,7 @@ import (
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
)
type URL struct {
@ -43,8 +44,8 @@ func (dma *StorageResource) Serialize() map[string]interface{} {
return m
}
func (d *StorageResource) GetAccessor() utils.Accessor {
func (d *StorageResource) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.STORAGE_RESOURCE)
data.Init(utils.STORAGE_RESOURCE, caller)
return data
}

6
models/resources/storage/storage_mongo_accessor.go
View File

@ -44,7 +44,7 @@ func (sma *storageMongoAccessor) LoadOne(id string) (utils.DBObject, int, error)
}
res_mongo.Decode(&storage)
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, sma.GetType())
if err == nil && len(resources) > 0 {
storage.ResourceModel = resources[0].(*resource_model.ResourceModel)
@ -63,7 +63,7 @@ func (wfa storageMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int, error)
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {
@ -96,7 +96,7 @@ func (wfa *storageMongoAccessor) Search(filters *dbs.Filters, search string) ([]
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {

5
models/resources/workflow/workflow.go
View File

@ -5,6 +5,7 @@ import (
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
)
type WorkflowResource struct {
@ -12,9 +13,9 @@ type WorkflowResource struct {
WorkflowID string `bson:"workflow_id,omitempty" json:"workflow_id,omitempty"`
}
func (d *WorkflowResource) GetAccessor() utils.Accessor {
func (d *WorkflowResource) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.WORKFLOW_RESOURCE)
data.Init(utils.WORKFLOW_RESOURCE, caller)
return data
}

6
models/resources/workflow/workflow_mongo_accessor.go
View File

@ -48,7 +48,7 @@ func (wfa *workflowResourceMongoAccessor) LoadOne(id string) (utils.DBObject, in
return nil, code, err
}
res_mongo.Decode(&workflow)
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
if err == nil && len(resources) > 0 {
workflow.ResourceModel = resources[0].(*resource_model.ResourceModel)
@ -67,7 +67,7 @@ func (wfa workflowResourceMongoAccessor) LoadAll() ([]utils.ShallowDBObject, int
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {
@ -100,7 +100,7 @@ func (wfa *workflowResourceMongoAccessor) Search(filters *dbs.Filters, search st
if err = res_mongo.All(mongo.MngoCtx, &results); err != nil {
return nil, 404, err
}
accessor := (&resource_model.ResourceModel{}).GetAccessor()
accessor := (&resource_model.ResourceModel{}).GetAccessor(nil)
resources, _, err := accessor.Search(nil, wfa.GetType())
for _, r := range results {
if err == nil && len(resources) > 0 {

11
models/utils/abstracts.go
View File

@ -7,6 +7,7 @@ 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/logs"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/go-playground/validator/v10"
"github.com/google/uuid"
"github.com/rs/zerolog"
@ -27,7 +28,7 @@ func (ao *AbstractObject) GetName() string {
return ao.Name
}
func (dma *AbstractObject) GetAccessor() Accessor {
func (dma *AbstractObject) GetAccessor(caller *tools.HTTPCaller) Accessor {
return nil
}
@ -57,14 +58,20 @@ func (r *AbstractObject) GenerateID() {
type AbstractAccessor struct {
Logger zerolog.Logger
Type string
Caller *tools.HTTPCaller
}
func (dma *AbstractAccessor) GetType() string {
return dma.Type
}
func (dma *AbstractAccessor) SetLogger(t DataType) {
func (dma *AbstractAccessor) GetCaller() *tools.HTTPCaller {
return dma.Caller
}
func (dma *AbstractAccessor) Init(t DataType, caller *tools.HTTPCaller) {
dma.Logger = logs.CreateLogger(t.String(), "")
dma.Caller = caller
dma.Type = t.String()
}

10
models/utils/interfaces.go
View File

@ -1,6 +1,9 @@
package utils
import "cloud.o-forge.io/core/oc-lib/dbs"
import (
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/tools"
)
type ShallowDBObject interface {
GenerateID()
@ -16,12 +19,13 @@ type DBObject interface {
GetName() string
Deserialize(j map[string]interface{}) DBObject
Serialize() map[string]interface{}
GetAccessor() Accessor
GetAccessor(caller *tools.HTTPCaller) Accessor
}
type Accessor interface {
SetLogger(t DataType)
Init(t DataType, caller *tools.HTTPCaller)
GetType() string
GetCaller() *tools.HTTPCaller
Search(filters *dbs.Filters, search string) ([]ShallowDBObject, int, error)
LoadAll() ([]ShallowDBObject, int, error)
LoadOne(id string) (DBObject, int, error)

73
models/workflow/workflow.go
View File

@ -2,14 +2,16 @@ package oclib
import (
"encoding/json"
"errors"
"fmt"
"slices"
"time"
"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/datacenter"
"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"
"cloud.o-forge.io/core/oc-lib/tools"
)
type AbstractWorkflow struct {
@ -34,9 +36,9 @@ type Workflow struct {
AbstractWorkflow
}
func (wfa *Workflow) CheckBooking() (bool, error) {
func (wfa *Workflow) CheckBooking(caller *tools.HTTPCaller) (bool, error) {
// check if
if wfa.Schedule == nil || wfa.Schedule.Start == nil {
if wfa.Schedule == nil || wfa.Schedule.Start == nil || wfa.Graph == nil {
return false, nil
}
if wfa.Schedule.End == nil {
@ -44,29 +46,60 @@ func (wfa *Workflow) CheckBooking() (bool, error) {
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
accessor := (&datacenter.DatacenterResource{}).GetAccessor(nil)
for _, link := range wfa.Graph.Links {
if ok, dc_id := wfa.isDCLink(link); ok {
dc, code, _ := accessor.LoadOne(dc_id)
if code != 200 {
continue
}
// CHECK BOOKING
url := dc.(*datacenter.DatacenterResource).SourceUrl
resp, err := caller.CallGet(url, caller.OriginSubPath+"/"+wfa.getFormat(wfa.Schedule.Start)+"/"+wfa.getFormat(&e))
if err != nil {
return false, err
}
var response map[string]interface{}
json.Unmarshal(resp, &response)
if code, ok := response["code"]; ok && code != 200 {
return false, errors.New(fmt.Sprintf("%v", response["error"]))
}
}
}
return len(res) == 0, nil
return true, nil
}
func (wfa *Workflow) getFormat(date *time.Time) string {
month := fmt.Sprintf("%v", date.Month())
day := fmt.Sprintf("%v", date.Day())
hour := fmt.Sprintf("%v", date.Hour())
minute := fmt.Sprintf("%v", date.Minute())
second := fmt.Sprintf("%v", date.Second())
if len(month) == 1 {
month = "0" + month
}
if len(day) == 1 {
day = "0" + day
}
if len(hour) == 1 {
hour = "0" + hour
}
if len(minute) == 1 {
minute = "0" + minute
}
if len(second) == 1 {
second = "0" + second
}
return fmt.Sprintf("%v", date.Year()) + "-" + month + "-" + day + "T" + hour + ":" + minute + ":" + second
}
func (d *Workflow) GetName() string {
return d.Name
}
func (d *Workflow) GetAccessor() utils.Accessor {
func (d *Workflow) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.WORKFLOW)
data.Init(utils.WORKFLOW, caller)
return data
}

68
models/workflow/workflow_mongo_accessor.go
View File

@ -1,12 +1,13 @@
package oclib
import (
"encoding/json"
"errors"
"fmt"
"strings"
"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"
@ -80,10 +81,9 @@ 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{}
func (wfa *workflowMongoAccessor) book(id string, realData *Workflow, execs []*workflow_execution.WorkflowExecution) error {
if realData.Schedule == nil {
return books
return nil
}
res, _, _ := wfa.LoadOne(id)
r := res.(*Workflow)
@ -92,46 +92,30 @@ func (wfa *workflowMongoAccessor) book(id string, realData *Workflow, execs []*w
g = realData.Graph
}
if g != nil && g.Links != nil && len(g.Links) > 0 {
bookAccessor := (&booking.Booking{}).GetAccessor()
accessor := (&datacenter.DatacenterResource{}).GetAccessor()
accessor := (&datacenter.DatacenterResource{}).GetAccessor(nil)
for _, link := range g.Links {
if ok, dc_id := realData.isDCLink(link); ok {
_, code, _ := accessor.LoadOne(dc_id)
dc, 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
url := dc.(*datacenter.DatacenterResource).SourceUrl
resp, err := wfa.Caller.CallPost(url, wfa.Caller.OriginSubPath, (&workflow_execution.WorkflowExecutions{
Executions: execs,
}).Serialize())
if err != nil {
return err
}
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))
}
var response map[string]interface{}
json.Unmarshal(resp, &response)
if code, ok := response["code"]; ok && code != 200 {
return errors.New(fmt.Sprintf("%v", response["error"]))
}
}
}
}
return books
return nil
}
func (wfa *workflowMongoAccessor) execution(id string, realData *Workflow, delete bool) (int, error) {
@ -143,15 +127,14 @@ 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
}
accessor := (&workflow_execution.WorkflowExecution{}).GetAccessor()
accessor := (&workflow_execution.WorkflowExecution{}).GetAccessor(nil)
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 err != nil {
return 422, err
}
err = wfa.book(id, realData, execs)
if err != nil {
return 409, err
}
if delete {
mongo.MONGOService.DeleteMultiple(map[string]interface{}{
@ -169,7 +152,6 @@ func (wfa *workflowMongoAccessor) execution(id string, realData *Workflow, delet
} else {
return 422, err
}
wfa.book(id, realData, execs)
return 200, nil
}
@ -202,7 +184,7 @@ func (wfa *workflowMongoAccessor) CopyOne(data utils.DBObject) (utils.DBObject,
}
func (wfa *workflowMongoAccessor) execute(workflow *Workflow) {
accessor := (&workspace.Workspace{}).GetAccessor()
accessor := (&workspace.Workspace{}).GetAccessor(nil)
filters := &dbs.Filters{
Or: map[string][]dbs.Filter{
"abstractobject.name": {{dbs.LIKE.String(), workflow.Name + "_workspace"}},

28
models/workflow_execution/workflow_execution.go
View File

@ -6,6 +6,7 @@ import (
"time"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
)
@ -38,6 +39,29 @@ func (d ScheduledType) EnumIndex() int {
return int(d)
}
type WorkflowExecutions struct {
Executions []*WorkflowExecution `json:"executions" bson:"executions"`
}
func (dma *WorkflowExecutions) Deserialize(j map[string]interface{}) *WorkflowExecutions {
b, err := json.Marshal(j)
if err != nil {
return nil
}
json.Unmarshal(b, dma)
return dma
}
func (dma *WorkflowExecutions) Serialize() map[string]interface{} {
var m map[string]interface{}
b, err := json.Marshal(dma)
if err != nil {
return nil
}
json.Unmarshal(b, &m)
return m
}
type WorkflowExecution struct {
utils.AbstractObject
ExecDate *time.Time `json:"execution_date,omitempty" bson:"execution_date,omitempty" validate:"required"`
@ -73,9 +97,9 @@ func (d *WorkflowExecution) GetName() string {
return d.UUID + "_" + d.ExecDate.String()
}
func (d *WorkflowExecution) GetAccessor() utils.Accessor {
func (d *WorkflowExecution) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.WORKFLOW_EXECUTION)
data.Init(utils.WORKFLOW_EXECUTION, caller)
return data
}

15
models/workspace/shared/rules/rule/rule.go
View File

@ -1,17 +1,18 @@
package rule
package rule
import (
"encoding/json"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
)
type Rule struct {
utils.AbstractObject
Description string `json:"description,omitempty" bson:"description,omitempty"`
Condition string `json:"condition,omitempty" bson:"condition,omitempty"`
Actions []string `json:"actions,omitempty" bson:"actions,omitempty"`
Description string `json:"description,omitempty" bson:"description,omitempty"`
Condition string `json:"condition,omitempty" bson:"condition,omitempty"`
Actions []string `json:"actions,omitempty" bson:"actions,omitempty"`
}
func (ao *Rule) GetID() string {
@ -26,9 +27,9 @@ func (d *Rule) GetName() string {
return d.Name
}
func (d *Rule) GetAccessor() utils.Accessor {
func (d *Rule) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.RULE)
data.Init(utils.RULE, caller)
return data
}

5
models/workspace/shared/shared_workspace.go
View File

@ -4,6 +4,7 @@ import (
"encoding/json"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
)
@ -30,9 +31,9 @@ func (d *SharedWorkspace) GetName() string {
return d.Name
}
func (d *SharedWorkspace) GetAccessor() utils.Accessor {
func (d *SharedWorkspace) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.SHARED_WORKSPACE)
data.Init(utils.SHARED_WORKSPACE, caller)
return data
}

4
models/workspace/shared/shared_workspace_mongo_accessor.go
View File

@ -28,7 +28,7 @@ func (wfa *sharedWorkspaceMongoAccessor) sharedWorkspace(shared *SharedWorkspace
eldest, code, _ := wfa.LoadOne(id)
if code == 200 {
eld := eldest.(*SharedWorkspace)
accessor := (&workspace.Workspace{}).GetAccessor()
accessor := (&workspace.Workspace{}).GetAccessor(nil)
if eld.Workspaces != nil {
for _, v := range eld.Workspaces {
accessor.UpdateOne(&workspace.Workspace{Shared: false}, v)
@ -43,7 +43,7 @@ func (wfa *sharedWorkspaceMongoAccessor) sharedWorkspace(shared *SharedWorkspace
}
func (wfa *sharedWorkspaceMongoAccessor) sharedWorkflow(shared *SharedWorkspace, id string) {
accessor := (&w.Workflow{}).GetAccessor()
accessor := (&w.Workflow{}).GetAccessor(nil)
eldest, code, _ := wfa.LoadOne(id)
if code == 200 {
eld := eldest.(*SharedWorkspace)

5
models/workspace/workspace.go
View File

@ -5,6 +5,7 @@ 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"
"github.com/google/uuid"
)
@ -28,9 +29,9 @@ func (d *Workspace) GetName() string {
return d.Name
}
func (d *Workspace) GetAccessor() utils.Accessor {
func (d *Workspace) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.SetLogger(utils.WORKSPACE)
data.Init(utils.WORKSPACE, caller)
return data
}

10
models/workspace/workspace_mongo_accessor.go
View File

@ -60,7 +60,7 @@ func (wfa *workspaceMongoAccessor) CopyOne(data utils.DBObject) (utils.DBObject,
func (wfa *workspaceMongoAccessor) fill(workflow *Workspace) *Workspace {
if workflow.Datas != nil && len(workflow.Datas) > 0 {
dataAccessor := (&data.DataResource{}).GetAccessor()
dataAccessor := (&data.DataResource{}).GetAccessor(nil)
for _, id := range workflow.Datas {
d, _, e := dataAccessor.LoadOne(id)
if e == nil {
@ -69,7 +69,7 @@ func (wfa *workspaceMongoAccessor) fill(workflow *Workspace) *Workspace {
}
}
if workflow.Datacenters != nil && len(workflow.Datacenters) > 0 {
dataAccessor := (&datacenter.DatacenterResource{}).GetAccessor()
dataAccessor := (&datacenter.DatacenterResource{}).GetAccessor(nil)
for _, id := range workflow.Datacenters {
d, _, e := dataAccessor.LoadOne(id)
if e == nil {
@ -78,7 +78,7 @@ func (wfa *workspaceMongoAccessor) fill(workflow *Workspace) *Workspace {
}
}
if workflow.Storages != nil && len(workflow.Storages) > 0 {
dataAccessor := (&storage.StorageResource{}).GetAccessor()
dataAccessor := (&storage.StorageResource{}).GetAccessor(nil)
for _, id := range workflow.Storages {
d, _, e := dataAccessor.LoadOne(id)
if e == nil {
@ -87,7 +87,7 @@ func (wfa *workspaceMongoAccessor) fill(workflow *Workspace) *Workspace {
}
}
if workflow.Processings != nil && len(workflow.Processings) > 0 {
dataAccessor := (&processing.ProcessingResource{}).GetAccessor()
dataAccessor := (&processing.ProcessingResource{}).GetAccessor(nil)
for _, id := range workflow.Processings {
d, _, e := dataAccessor.LoadOne(id)
if e == nil {
@ -96,7 +96,7 @@ func (wfa *workspaceMongoAccessor) fill(workflow *Workspace) *Workspace {
}
}
if workflow.Workflows != nil && len(workflow.Workflows) > 0 {
dataAccessor := (&w.WorkflowResource{}).GetAccessor()
dataAccessor := (&w.WorkflowResource{}).GetAccessor(nil)
for _, id := range workflow.Workflows {
d, _, e := dataAccessor.LoadOne(id)
if e == nil {

44
tools/remote_caller.go Normal file
View File

@ -0,0 +1,44 @@
package tools
import (
"bytes"
"encoding/json"
"io"
"net/http"
)
var HTTPCallerInstance = &HTTPCaller{}
type HTTPCaller struct {
Origin string
OriginSubPath string
DestSubPath string
}
func NewHTTPCaller(url string, origin string, originSubPath string, destSubPath string) *HTTPCaller {
return &HTTPCaller{
Origin: origin,
OriginSubPath: originSubPath,
DestSubPath: destSubPath,
}
}
func (caller *HTTPCaller) CallGet(url string, subpath string) ([]byte, error) {
resp, err := http.Get(url + "/" + subpath)
if err != nil {
return nil, err
}
defer resp.Body.Close()
return io.ReadAll(resp.Body)
}
func (caller *HTTPCaller) CallPost(url string, subpath string, body map[string]interface{}) ([]byte, error) {
postBody, _ := json.Marshal(body)
responseBody := bytes.NewBuffer(postBody)
resp, err := http.Post(url+"/"+subpath, "application/json", responseBody)
if err != nil {
return nil, err
}
defer resp.Body.Close()
return io.ReadAll(resp.Body)
}