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

A question refers to the comment ! And if not Ooopsy

Browse Source
This commit is contained in:
mr
2024-08-30 14:50:48 +02:00
parent db78c70dc3
commit 8180fe5e99
39 changed files with 737 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
models/peer/peer.go
View File

@@ -10,17 +10,19 @@ import (
"github.com/google/uuid"
)
// Peer is a struct that represents a peer
type Peer struct {
utils.AbstractObject
Url string `json:"url,omitempty" bson:"url,omitempty" validate:"required,base64url"`
PublicKey string `json:"public_key,omitempty" bson:"public_key,omitempty"`
Services map[string]int `json:"services,omitempty" bson:"services,omitempty"`
FailedExecution []PeerExecution `json:"failed_execution" bson:"failed_execution"`
Url string `json:"url,omitempty" bson:"url,omitempty" validate:"required,base64url"` // Url is the URL of the peer (base64url)
PublicKey string `json:"public_key,omitempty" bson:"public_key,omitempty"` // PublicKey is the public key of the peer
Services map[string]int `json:"services,omitempty" bson:"services,omitempty"` // Services is the services of the peer
FailedExecution []PeerExecution `json:"failed_execution" bson:"failed_execution"` // FailedExecution is the list of failed executions, to be retried
}
// AddExecution adds an execution to the list of failed executions
func (ao *Peer) AddExecution(exec PeerExecution) {
found := false
for _, v := range ao.FailedExecution {
for _, v := range ao.FailedExecution { // Check if the execution is already in the list
if v.Url == exec.Url && v.Method == exec.Method && fmt.Sprint(v.Body) == fmt.Sprint(exec.Body) {
found = true
break
@@ -31,6 +33,7 @@ func (ao *Peer) AddExecution(exec PeerExecution) {
}
}
// RemoveExecution removes an execution from the list of failed executions
func (ao *Peer) RemoveExecution(exec PeerExecution) {
new := []PeerExecution{}
for i, v := range ao.FailedExecution {
@@ -41,14 +44,16 @@ func (ao *Peer) RemoveExecution(exec PeerExecution) {
ao.FailedExecution = new
}
// IsMySelf checks if the peer is the local peer
func (ao *Peer) IsMySelf() bool {
id, _ := static.GetMyLocalJsonPeer()
return ao.UUID == id
}
// LaunchPeerExecution launches an execution on a peer
func (p *Peer) LaunchPeerExecution(peerID string, dataID string, dt utils.DataType, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) (*PeerExecution, error) {
p.UUID = peerID
return (&PeerCache{}).LaunchPeerExecution(peerID, dataID, dt, method, body, caller)
return cache.LaunchPeerExecution(peerID, dataID, dt, method, body, caller) // Launch the execution on the peer through the cache
}
func (ao *Peer) GetID() string {
@@ -64,8 +69,8 @@ func (d *Peer) GetName() string {
}
func (d *Peer) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.Init(utils.PEER, caller)
data := New() // Create a new instance of the accessor
data.Init(utils.PEER, caller) // Initialize the accessor with the PEER model type
return data
}

74
models/peer/peer_cache.go
View File

@@ -11,6 +11,10 @@ import (
"cloud.o-forge.io/core/oc-lib/tools"
)
/*
* PeerExecution is a struct that represents an execution on a peer
* it defines the execution data
*/
type PeerExecution struct {
Method string `json:"method" bson:"method"`
Url string `json:"url" bson:"url"`
@@ -19,11 +23,16 @@ type PeerExecution struct {
DataID string `json:"data_id" bson:"data_id"`
}
var cache = &PeerCache{} // Singleton instance of the peer cache
// PeerCache is a struct that represents a peer cache
type PeerCache struct {
Executions []*PeerExecution
}
// urlFormat formats the URL of the peer with the data type API function
func (p *PeerCache) urlFormat(url string, dt utils.DataType) string {
// localhost is replaced by the local peer URL
// because localhost must collide on a web request security protocol
if strings.Contains(url, "localhost") || strings.Contains(url, "127.0.0.1") {
url = strings.ReplaceAll(url, "localhost", dt.API())
url = strings.ReplaceAll(url, "127.0.0.1", dt.API())
@@ -37,53 +46,46 @@ func (p *PeerCache) urlFormat(url string, dt utils.DataType) string {
return url
}
// checkPeerStatus checks the status of a peer
func (p *PeerCache) checkPeerStatus(peerID string, appName string, caller *tools.HTTPCaller) (*Peer, bool) {
api := tools.API{}
access := (&Peer{}).GetAccessor(nil)
res, code, _ := access.LoadOne(peerID)
if code != 200 {
res, code, _ := access.LoadOne(peerID) // Load the peer from db
if code != 200 { // no peer no party
return nil, false
}
methods := caller.URLS[utils.PEER.String()]
fmt.Println("PEER AFT 3", methods)
methods := caller.URLS[utils.PEER.String()] // Get the methods url of the peer
if methods == nil {
return res.(*Peer), false
}
meth := methods[tools.POST]
fmt.Println("PEER AFT 4", meth)
meth := methods[tools.POST] // Get the POST method to check status
if meth == "" {
return res.(*Peer), false
}
url := p.urlFormat(res.(*Peer).Url+meth, utils.PEER)
fmt.Println("PEER AFT 5", url)
state, services := api.CheckRemotePeer(url)
res.(*Peer).Services = services
access.UpdateOne(res, peerID)
fmt.Println("PEER AFT 6", state, services, appName)
return res.(*Peer), state != tools.DEAD && services[appName] == 0
}
func (p *PeerCache) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.Init(utils.PEER, caller)
return data
url := p.urlFormat(res.(*Peer).Url+meth, utils.PEER) // Format the URL
state, services := api.CheckRemotePeer(url) // Check the status of the peer
res.(*Peer).Services = services // Update the services states of the peer
access.UpdateOne(res, peerID) // Update the peer in the db
return res.(*Peer), state != tools.DEAD && services[appName] == 0 // Return the peer and its status
}
// LaunchPeerExecution launches an execution on a peer
func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string,
dt utils.DataType, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) (*PeerExecution, error) {
methods := caller.URLS[dt.String()]
methods := caller.URLS[dt.String()] // Get the methods url of the data type
if _, ok := methods[method]; !ok {
return nil, errors.New("no path found")
}
meth := methods[method]
meth := methods[method] // Get the method url to execute
if meth == "" {
return nil, errors.New("no path found")
} else {
meth = strings.ReplaceAll(meth, ":id", dataID)
meth = strings.ReplaceAll(meth, ":id", dataID) // Replace the id in the url in case of a DELETE / UPDATE method (it's a standard naming in OC)
}
url := ""
// Check the status of the peer
if mypeer, ok := p.checkPeerStatus(peerID, dt.API(), caller); !ok {
// 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),
@@ -92,31 +94,32 @@ func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string,
DataID: dataID,
}
mypeer.AddExecution(*pexec)
mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID)
mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID) // Update the peer in the db
return nil, errors.New("peer is not reachable")
} else {
url = p.urlFormat((mypeer.Url)+meth, dt)
fmt.Println("LaunchPeerExecution AFT 3", url)
tmp := mypeer.FailedExecution
mypeer.FailedExecution = []PeerExecution{}
mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID)
for _, v := range tmp {
// If the peer is reachable, launch the execution
url = p.urlFormat((mypeer.Url)+meth, dt) // Format the URL
tmp := mypeer.FailedExecution // Get the failed executions list
mypeer.FailedExecution = []PeerExecution{} // Reset the failed executions list
mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID) // Update the peer in the db
for _, v := range tmp { // Retry the failed executions
go p.exec(v.Url, tools.ToMethod(v.Method), v.Body, caller)
}
}
return nil, p.exec(url, method, body, caller)
return nil, p.exec(url, method, body, caller) // Execute the method
}
// exec executes the method on the peer
func (p *PeerCache) exec(url string, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) error {
var b []byte
var err error
if method == tools.POST {
if method == tools.POST { // Execute the POST method if it's a POST method
b, err = caller.CallPost(url, "", body)
}
if method == tools.GET {
if method == tools.GET { // Execute the GET method if it's a GET method
b, err = caller.CallGet(url, "")
}
if method == tools.DELETE {
if method == tools.DELETE { // Execute the DELETE method if it's a DELETE method
b, err = caller.CallDelete(url, "")
}
var m map[string]interface{}
@@ -124,9 +127,8 @@ func (p *PeerCache) exec(url string, method tools.METHOD, body map[string]interf
if err != nil {
return err
}
fmt.Println("LaunchPeerExecution AFT 3", m, url)
if e, ok := m["error"]; !ok && e != "" {
if e, ok := m["error"]; !ok && e != "" { // Check if there is an error in the response
return errors.New(fmt.Sprintf("%v", m["error"]))
}
return err

9
models/peer/peer_mongo_accessor.go
View File

@@ -7,13 +7,18 @@ import (
)
type peerMongoAccessor struct {
utils.AbstractAccessor
utils.AbstractAccessor // AbstractAccessor contains the basic fields of an accessor (model, caller)
}
// New creates a new instance of the peerMongoAccessor
func New() *peerMongoAccessor {
return &peerMongoAccessor{}
}
/*
* Nothing special here, just the basic CRUD operations
*/
func (wfa *peerMongoAccessor) DeleteOne(id string) (utils.DBObject, int, error) {
return wfa.GenericDeleteOne(id, wfa)
}
@@ -63,7 +68,7 @@ func (wfa *peerMongoAccessor) Search(filters *dbs.Filters, search string) ([]uti
objs := []utils.ShallowDBObject{}
if (filters == nil || len(filters.And) == 0 || len(filters.Or) == 0) && search != "" {
filters = &dbs.Filters{
Or: map[string][]dbs.Filter{
Or: map[string][]dbs.Filter{ // search by name if no filters are provided
"abstractobject.name": {{Operator: dbs.LIKE.String(), Value: search}},
},
}