2024-08-13 14:33:26 +02:00
|
|
|
package peer
|
|
|
|
|
|
|
|
import (
|
|
|
|
"encoding/json"
|
|
|
|
"errors"
|
|
|
|
"fmt"
|
2024-08-22 10:19:14 +02:00
|
|
|
"regexp"
|
2024-08-13 14:33:26 +02:00
|
|
|
"strings"
|
|
|
|
|
|
|
|
"cloud.o-forge.io/core/oc-lib/models/utils"
|
|
|
|
"cloud.o-forge.io/core/oc-lib/tools"
|
|
|
|
)
|
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
/*
|
|
|
|
* PeerExecution is a struct that represents an execution on a peer
|
|
|
|
* it defines the execution data
|
|
|
|
*/
|
2024-08-13 14:33:26 +02:00
|
|
|
type PeerExecution struct {
|
2024-08-23 09:53:37 +02:00
|
|
|
Method string `json:"method" bson:"method"`
|
|
|
|
Url string `json:"url" bson:"url"`
|
|
|
|
Body map[string]interface{} `json:"body" bson:"body"`
|
|
|
|
DataType int `json:"data_type" bson:"data_type"`
|
|
|
|
DataID string `json:"data_id" bson:"data_id"`
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
var cache = &PeerCache{} // Singleton instance of the peer cache
|
|
|
|
// PeerCache is a struct that represents a peer cache
|
2024-08-13 14:33:26 +02:00
|
|
|
type PeerCache struct {
|
|
|
|
Executions []*PeerExecution
|
|
|
|
}
|
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
// urlFormat formats the URL of the peer with the data type API function
|
2024-08-22 16:13:07 +02:00
|
|
|
func (p *PeerCache) urlFormat(url string, dt utils.DataType) string {
|
2024-08-30 14:50:48 +02:00
|
|
|
// localhost is replaced by the local peer URL
|
|
|
|
// because localhost must collide on a web request security protocol
|
2024-08-22 15:18:59 +02:00
|
|
|
if strings.Contains(url, "localhost") || strings.Contains(url, "127.0.0.1") {
|
2024-08-22 16:13:07 +02:00
|
|
|
url = strings.ReplaceAll(url, "localhost", dt.API())
|
|
|
|
url = strings.ReplaceAll(url, "127.0.0.1", dt.API())
|
2024-08-22 15:18:59 +02:00
|
|
|
r := regexp.MustCompile("(:[0-9]+)")
|
|
|
|
t := r.FindString(url)
|
|
|
|
if t != "" {
|
|
|
|
url = strings.Replace(url, t, ":8080", -1)
|
|
|
|
}
|
|
|
|
r.ReplaceAllString(url, ":8080")
|
|
|
|
}
|
|
|
|
return url
|
|
|
|
}
|
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
// checkPeerStatus checks the status of a peer
|
2024-08-23 10:01:37 +02:00
|
|
|
func (p *PeerCache) checkPeerStatus(peerID string, appName string, caller *tools.HTTPCaller) (*Peer, bool) {
|
2024-08-21 15:46:16 +02:00
|
|
|
api := tools.API{}
|
|
|
|
access := (&Peer{}).GetAccessor(nil)
|
2024-08-30 14:50:48 +02:00
|
|
|
res, code, _ := access.LoadOne(peerID) // Load the peer from db
|
|
|
|
if code != 200 { // no peer no party
|
2024-08-23 09:53:37 +02:00
|
|
|
return nil, false
|
2024-08-21 15:46:16 +02:00
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
methods := caller.URLS[utils.PEER.String()] // Get the methods url of the peer
|
2024-08-21 16:03:56 +02:00
|
|
|
if methods == nil {
|
2024-08-23 09:53:37 +02:00
|
|
|
return res.(*Peer), false
|
2024-08-21 16:03:56 +02:00
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
meth := methods[tools.POST] // Get the POST method to check status
|
2024-08-21 16:03:56 +02:00
|
|
|
if meth == "" {
|
2024-08-23 09:53:37 +02:00
|
|
|
return res.(*Peer), false
|
2024-08-21 16:03:56 +02:00
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
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
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
// LaunchPeerExecution launches an execution on a peer
|
2024-08-23 09:53:37 +02:00
|
|
|
func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string,
|
2024-08-21 08:54:29 +02:00
|
|
|
dt utils.DataType, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) (*PeerExecution, error) {
|
2024-08-30 14:50:48 +02:00
|
|
|
methods := caller.URLS[dt.String()] // Get the methods url of the data type
|
2024-08-13 14:33:26 +02:00
|
|
|
if _, ok := methods[method]; !ok {
|
|
|
|
return nil, errors.New("no path found")
|
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
meth := methods[method] // Get the method url to execute
|
2024-08-22 16:28:21 +02:00
|
|
|
if meth == "" {
|
|
|
|
return nil, errors.New("no path found")
|
|
|
|
} else {
|
2024-08-30 14:50:48 +02:00
|
|
|
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)
|
2024-08-22 16:28:21 +02:00
|
|
|
}
|
2024-08-23 09:53:37 +02:00
|
|
|
url := ""
|
2024-08-30 14:50:48 +02:00
|
|
|
// Check the status of the peer
|
2024-08-23 10:01:37 +02:00
|
|
|
if mypeer, ok := p.checkPeerStatus(peerID, dt.API(), caller); !ok {
|
2024-08-30 14:50:48 +02:00
|
|
|
// If the peer is not reachable, add the execution to the failed executions list
|
2024-08-23 23:00:57 +02:00
|
|
|
pexec := &PeerExecution{
|
|
|
|
Method: method.String(),
|
|
|
|
Url: p.urlFormat((mypeer.Url)+meth, dt),
|
|
|
|
Body: body,
|
|
|
|
DataType: dt.EnumIndex(),
|
|
|
|
DataID: dataID,
|
|
|
|
}
|
2024-08-23 09:53:37 +02:00
|
|
|
mypeer.AddExecution(*pexec)
|
2024-08-30 14:50:48 +02:00
|
|
|
mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID) // Update the peer in the db
|
2024-08-22 12:29:38 +02:00
|
|
|
return nil, errors.New("peer is not reachable")
|
2024-08-23 09:53:37 +02:00
|
|
|
} else {
|
2024-08-30 14:50:48 +02:00
|
|
|
// 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
|
2024-08-23 23:00:57 +02:00
|
|
|
go p.exec(v.Url, tools.ToMethod(v.Method), v.Body, caller)
|
2024-08-23 09:53:37 +02:00
|
|
|
}
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
return nil, p.exec(url, method, body, caller) // Execute the method
|
2024-08-23 23:00:57 +02:00
|
|
|
}
|
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
// exec executes the method on the peer
|
2024-08-23 23:00:57 +02:00
|
|
|
func (p *PeerCache) exec(url string, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) error {
|
|
|
|
var b []byte
|
|
|
|
var err error
|
2024-09-23 10:12:52 +02:00
|
|
|
fmt.Println("executing", url, method, body)
|
2024-08-30 14:50:48 +02:00
|
|
|
if method == tools.POST { // Execute the POST method if it's a POST method
|
2024-08-22 16:28:21 +02:00
|
|
|
b, err = caller.CallPost(url, "", body)
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
if method == tools.GET { // Execute the GET method if it's a GET method
|
2024-08-22 16:28:21 +02:00
|
|
|
b, err = caller.CallGet(url, "")
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
2024-08-30 14:50:48 +02:00
|
|
|
if method == tools.DELETE { // Execute the DELETE method if it's a DELETE method
|
2024-08-22 16:28:21 +02:00
|
|
|
b, err = caller.CallDelete(url, "")
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
|
|
|
var m map[string]interface{}
|
|
|
|
json.Unmarshal(b, &m)
|
2024-09-23 10:12:52 +02:00
|
|
|
fmt.Println("response", err)
|
2024-08-13 14:33:26 +02:00
|
|
|
if err != nil {
|
2024-08-23 23:00:57 +02:00
|
|
|
return err
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
2024-08-22 08:50:24 +02:00
|
|
|
|
2024-08-30 14:50:48 +02:00
|
|
|
if e, ok := m["error"]; !ok && e != "" { // Check if there is an error in the response
|
2024-08-23 23:00:57 +02:00
|
|
|
return errors.New(fmt.Sprintf("%v", m["error"]))
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|
2024-08-23 23:00:57 +02:00
|
|
|
return err
|
2024-08-13 14:33:26 +02:00
|
|
|
}
|