oc-lib/models/peer/peer_cache.go

package peer

import (
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strings"

	"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"`
	Body     interface{} `json:"body" bson:"body"`
	DataType int         `json:"data_type" bson:"data_type"`
	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(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(hostUrl, "localhost") {
		localhost = "localhost"
	}
	if strings.Contains(hostUrl, "127.0.0.1") {
		localhost = "127.0.0.1"
	}
	if localhost != "" {
		r := regexp.MustCompile("(" + localhost + ":[0-9]+)")
		t := r.FindString(hostUrl)
		if t != "" {
			hostUrl = strings.Replace(hostUrl, t, dt.API()+":8080/oc", -1)
		} else {
			hostUrl = strings.ReplaceAll(hostUrl, localhost, dt.API()+":8080/oc")
		}
	} else {
		hostUrl = hostUrl + "/" + strings.ReplaceAll(dt.API(), "oc-", "")
	}
	return hostUrl
}

// checkPeerStatus checks the status of a peer
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
	}
	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
	return res.(*Peer), state != tools.DEAD && services[appName] == 0 // Return the peer and its status
}

// LaunchPeerExecution launches an execution on a peer
// The method contacts the path described by : peer.Url + datatype path (from enums) + replacement of id by dataID
func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string,
	dt tools.DataType, method tools.METHOD, body interface{}, caller *tools.HTTPCaller) (*PeerExecution, error) {
	fmt.Println("Launching peer execution on", caller.URLS, dt, method)
	methods := caller.URLS[dt] // Get the methods url of the data type
	if m, ok := methods[method]; !ok || m == "" {
		return nil, errors.New("Requested method " + method.String() + " not declared in HTTPCaller")
	}
	path := methods[method]                        // Get the path corresponding to the action we want to execute
	path = strings.ReplaceAll(path, ":id", dataID) // Replace the id in the path 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()); !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), dt) + path,  // the url is constitued of : host URL + resource path + action path (ex : mypeer.com/datacenter/resourcetype/path/to/action)
			Body:     body,
			DataType: dt.EnumIndex(),
			DataID:   dataID,
		}
		mypeer.AddExecution(*pexec)
		NewShallowAccessor().UpdateOne(mypeer, peerID) // Update the peer in the db
		return nil, errors.New("peer is not reachable")
	} else {
		if mypeer == nil {
			return nil, errors.New("peer not found")
		}
		// If the peer is reachable, launch the execution
		url = p.urlFormat((mypeer.Url), dt) + path       // 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
		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) // Execute the method
}

// exec executes the method on the peer
func (p *PeerCache) exec(url string, method tools.METHOD, body interface{}, caller *tools.HTTPCaller) error {
	var b []byte
	var err error
	if method == tools.POST { // Execute the POST method if it's a POST method
		b, err = caller.CallPost(url, "", body)
	}
	if method == tools.GET { // Execute the GET method if it's a GET method
		b, err = caller.CallGet(url, "")
	}
	if method == tools.DELETE { // Execute the DELETE method if it's a DELETE method
		b, err = caller.CallDelete(url, "")
	}
	if err != nil {
		return err
	}
	var m map[string]interface{}
	err = json.Unmarshal(b, &m)
	if err != nil {
		return err
	}
	if e, ok := m["error"]; ok && e != "<nil>" && e != "" { // Check if there is an error in the response
		return errors.New(fmt.Sprintf("%v", m["error"]))
	}
	return nil
}