Automatic offloading target discovery

Author: grussorusso

docs/registry.md

@@ -0,0 +1,24 @@
+## Registry
+
+### Node Registration
+
+Each node is configured to belong to a certain geographical **Area** (e.g.,
+an Edge zone in Rome, a Cloud region in London).
+At startup, a (likely) unique random **Key** is associated with each node.
+Together, area and key represent a node's identifier.
+
+Each node registers itself in Etcd under `registry/<area>/<key>`. The value
+associated with this Etcd key is a string `apiPort;udpPort`, which reports the 
+port numbers used by the node for the API server and the UDP local monitoring
+server.
+
+
+### Load Balancer
+
+Load balancer nodes register themselves under `registry/<area>/lb/<key>`.
+
+
+
+
+
+

internal/api/api.go

@@ -240,7 +240,6 @@ func GetServerStatus(c echo.Context) error {
 
 	// TODO: use a different type
 	response := registration.StatusInformation{
-		Url:                     registration.SelfRegistration.RemoteURL,
 		AvailableWarmContainers: node.WarmStatus(),
 		AvailableMemMB:          node.Resources.AvailableMemMB,
 		AvailableCPUs:           node.Resources.AvailableCPUs,

internal/config/keys.go

@@ -7,9 +7,6 @@ const ETCD_ADDRESS = "etcd.address"
 const API_PORT = "api.port"
 const API_IP = "api.ip"
 
-// REMOTE SERVER URL
-const CLOUD_URL = "cloud.server.url"
-
 // Forces runtime container images to be pulled the first time they are used,
 // even if they are locally available (true/false).
 const FACTORY_REFRESH_IMAGES = "factory.images.refresh"

internal/lb/lb.go

@@ -42,7 +42,7 @@ func StartReverseProxy(e *echo.Echo, region string) {
 }
 
 func getTargets(region string) ([]*middleware.ProxyTarget, error) {
-	cloudNodes, err := registration.GetAllInArea(region, false)
+	cloudNodes, err := registration.GetNodesInArea(region, false, 0)
 	if err != nil {
 		return nil, err
 	}
@@ -51,7 +51,7 @@ func getTargets(region string) ([]*middleware.ProxyTarget, error) {
 	for _, target := range cloudNodes {
 		log.Printf("Found target: %v\n", target.Key)
 		// TODO: etcd should NOT contain URLs, but only host and port...
-		parsedUrl, err := url.Parse(target.RemoteURL)
+		parsedUrl, err := url.Parse(target.APIUrl())
 		if err != nil {
 			return nil, err
 		}

internal/node/node.go

@@ -19,7 +19,7 @@ type NodeID struct {
 var LocalNode NodeID
 
 func (n NodeID) String() string {
-	return fmt.Sprintf("%s/%s", n.Area, n.Key)
+	return fmt.Sprintf("(%s)%s", n.Area, n.Key)
 }
 
 func NewIdentifier(area string) NodeID {

internal/registration/registry.go

@@ -6,9 +6,10 @@ import (
 	"github.com/serverledge-faas/serverledge/internal/node"
 	"golang.org/x/exp/maps"
 	"log"
-	"net/url"
 	"path"
 	"sort"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -19,14 +20,15 @@ import (
 )
 
 const registryBaseDirectory = "registry"
-const registryLoadBalancerDirectory = "lb"
+const registryLoadBalancerDirectory = "__lb"
 const etcdLeaseTTL = 120
 
 var mutex sync.RWMutex
 
 var nearestNeighbors []NodeRegistration
 var neighborInfo map[string]*StatusInformation
 var neighbors map[string]NodeRegistration
+var remoteOffloadingTarget NodeRegistration
 
 var VivaldiClient *vivaldi.Client
 var SelfRegistration *NodeRegistration
@@ -42,6 +44,10 @@ func (r *NodeRegistration) toEtcdKey() (key string) {
 	}
 }
 
+func (r *NodeRegistration) APIUrl() (url string) {
+	return fmt.Sprintf("http://%s:%d", r.IPAddress, r.APIPort)
+}
+
 func areaEtcdKey(area string) string {
 	return fmt.Sprintf("%s/%s/", registryBaseDirectory, area)
 }
@@ -71,14 +77,17 @@ func registerToEtcd(asLoadBalancer bool) error {
 	etcdLease = resp.ID
 
 	registeredLocalIP := config.GetString(config.API_IP, defaultAddressStr)
-	hostport := fmt.Sprintf("http://%s:%d", registeredLocalIP, config.GetInt(config.API_PORT, 1323))
+	apiPort := config.GetInt(config.API_PORT, 1323)
+	udpPort := config.GetInt(config.LISTEN_UDP_PORT, 9876)
 
-	SelfRegistration = &NodeRegistration{NodeID: node.LocalNode, IPAddress: registeredLocalIP, RemoteURL: hostport, IsLoadBalancer: asLoadBalancer}
+	payload := fmt.Sprintf("%d;%d", apiPort, udpPort)
+
+	SelfRegistration = &NodeRegistration{NodeID: node.LocalNode, IPAddress: registeredLocalIP, APIPort: apiPort, UDPPort: udpPort, IsLoadBalancer: asLoadBalancer}
 
 	// save couple (id, hostport) to the correct Area-dir on etcd
 	etcdKey := SelfRegistration.toEtcdKey()
 	log.Printf("Registering to etcd: %s\n", etcdKey)
-	_, err = etcdClient.Put(ctx, etcdKey, hostport, clientv3.WithLease(etcdLease))
+	_, err = etcdClient.Put(ctx, etcdKey, payload, clientv3.WithLease(etcdLease))
 	if err != nil {
 		log.Fatal(IdRegistrationErr)
 		return IdRegistrationErr
@@ -109,36 +118,73 @@ func keepAliveLease() {
 	}
 }
 
-// GetAllInArea is used to obtain the list of  other server's addresses under a specific local Area
-func GetAllInArea(area string, includeSelf bool) (map[string]NodeRegistration, error) {
+func parseEtcdRegisteredNode(area string, key string, payload []byte) (NodeRegistration, error) {
+	payloadStr := string(payload)
+	split := strings.Split(payloadStr, ";")
+	if len(split) < 2 {
+		return NodeRegistration{}, fmt.Errorf("invalid payload: %s", payloadStr)
+	}
+
+	apiPort, err := strconv.Atoi(split[0])
+	if err != nil {
+		return NodeRegistration{}, err
+	}
+
+	udpPort, err := strconv.Atoi(split[1])
+	if err != nil {
+		return NodeRegistration{}, err
+	}
+
+	return NodeRegistration{NodeID: node.NodeID{Area: area, Key: key}, APIPort: apiPort, UDPPort: udpPort}, nil
+}
+
+// GetNodesInArea is used to obtain the list of  other server's addresses under a specific local Area
+func GetNodesInArea(area string, includeSelf bool, limit int64) (map[string]NodeRegistration, error) {
 	baseDir := areaEtcdKey(area)
+	lbPrefix := path.Join(baseDir, registryLoadBalancerDirectory)
 
 	ctx, _ := context.WithTimeout(context.Background(), 3*time.Second)
 
-	resp, err := etcdClient.Get(ctx, baseDir, clientv3.WithPrefix())
+	if limit < 0 {
+		limit = 0 // no limit
+	}
+	resp, err := etcdClient.Get(ctx, baseDir, clientv3.WithPrefix(), clientv3.WithLimit(limit))
 	if err != nil {
 		return nil, fmt.Errorf("Could not read from etcd: %v", err)
 	}
 
 	servers := make(map[string]NodeRegistration)
 	for _, s := range resp.Kvs {
-		key := path.Base(string(s.Key))
-		if key == registryLoadBalancerDirectory {
-			// TODO: is this check needed?
+		if strings.HasPrefix(string(s.Key), lbPrefix) {
+			// skip LB
 			continue
 		}
+		key := path.Base(string(s.Key))
 		if !includeSelf && area == SelfRegistration.Area && key == SelfRegistration.Key {
 			continue
 		}
 
-		remoteURL := string(s.Value)
-		servers[key] = NodeRegistration{NodeID: node.NodeID{Area: area, Key: key}, RemoteURL: remoteURL}
-		fmt.Printf("Server found: %v\n", servers[key])
+		reg, err := parseEtcdRegisteredNode(area, key, s.Value)
+		if err == nil {
+			servers[key] = reg
+			fmt.Printf("Server found: %v\n", servers[key])
+		}
 	}
 
 	return servers, nil
 }
 
+func GetOneNodeInArea(area string, includeSelf bool) (NodeRegistration, error) {
+	nodes, err := GetNodesInArea(area, includeSelf, 1)
+	if err == nil {
+		for _, n := range nodes {
+			return n, nil
+		}
+	}
+
+	return NodeRegistration{}, err
+}
+
 func GetLBInArea(area string) (map[string]NodeRegistration, error) {
 	baseDir := areaEtcdKey(area) + "/" + registryLoadBalancerDirectory
 
@@ -152,26 +198,23 @@ func GetLBInArea(area string) (map[string]NodeRegistration, error) {
 	servers := make(map[string]NodeRegistration)
 	for _, s := range resp.Kvs {
 		key := path.Base(string(s.Key))
-		remoteURL := string(s.Value)
-		servers[key] = NodeRegistration{NodeID: node.NodeID{Area: area, Key: key}, RemoteURL: remoteURL, IsLoadBalancer: true}
-		fmt.Printf("Server found: %v\n", servers[key])
+		reg, err := parseEtcdRegisteredNode(area, key, s.Value)
+		if err == nil {
+			reg.IsLoadBalancer = true
+			servers[key] = reg
+			fmt.Printf("Server found: %v\n", servers[key])
+		}
 	}
 
 	return servers, nil
 }
 
 // Deregister deletes from etcd the key, value pair previously inserted
 func Deregister() (e error) {
-	etcdClient, err := utils.GetEtcdClient()
-	if err != nil {
-		log.Fatal(UnavailableClientErr)
-		return UnavailableClientErr
-	}
-
 	ctx, _ := context.WithTimeout(context.Background(), 1*time.Second)
-	_, err = etcdClient.Revoke(ctx, etcdLease)
+	_, err := etcdClient.Revoke(ctx, etcdLease)
 	if err != nil {
-		return err
+		log.Printf("Error revoking lease: %v", err)
 	}
 
 	return nil
@@ -217,7 +260,7 @@ func runMonitor() {
 func globalMonitoring() {
 
 	// gets info from Etcd about other nodes in the area
-	newNeighbors, err := GetAllInArea(SelfRegistration.Area, false)
+	newNeighbors, err := GetNodesInArea(SelfRegistration.Area, false, 0)
 	if err != nil {
 		log.Println(err)
 		return
@@ -235,6 +278,38 @@ func globalMonitoring() {
 			delete(neighborInfo, key)
 		}
 	}
+
+	updateRemoteOffloadingTarget()
+}
+
+func updateRemoteOffloadingTarget() {
+	// If there is a LB in the remote area, it is used.
+	// Otherwise, a random node in the area is chosen.
+
+	remoteArea := config.GetString(config.REGISTRY_REMOTE_AREA, "")
+	if remoteArea == "" {
+		log.Printf("No remote area is configured; vertical offloading disabled")
+		remoteOffloadingTarget = NodeRegistration{}
+		return
+	}
+
+	lbs, err := GetLBInArea(remoteArea)
+	if err != nil {
+		log.Println(err)
+	}
+	if err == nil && len(lbs) > 0 {
+		for _, lb := range lbs {
+			log.Printf("Using LB as offloading target: %v", lb.NodeID)
+			remoteOffloadingTarget = lb
+			return
+		}
+	}
+
+	remoteNode, err := GetOneNodeInArea(remoteArea, false)
+	if err == nil {
+		log.Printf("Using as offloading target: %v", remoteNode.NodeID)
+		remoteOffloadingTarget = remoteNode
+	}
 }
 
 // computeNearestNeighbors finds servers nearby to the current one
@@ -278,22 +353,17 @@ func nearbyMonitoring(vivaldiClient *vivaldi.Client) {
 	mutex.RUnlock()
 
 	for _, registeredNode := range peersToUpdate {
-		u, err := url.Parse(registeredNode.RemoteURL)
-		if err != nil {
-			panic(err)
-		}
-		hostname := u.Hostname()
-		newInfo, rtt := statusInfoRequest(hostname)
+		newInfo, rtt := statusInfoRequest(&registeredNode)
 
 		if newInfo == nil {
-			log.Printf("Unreachable neighbor: %s\n", registeredNode.RemoteURL)
+			log.Printf("Unreachable neighbor: %s\n", registeredNode.NodeID)
 			continue
 		}
 
 		mutex.Lock()
 		neighborInfo[registeredNode.Key] = newInfo
 
-		_, err = vivaldiClient.Update("node", &newInfo.Coordinates, rtt)
+		_, err := vivaldiClient.Update("node", &newInfo.Coordinates, rtt)
 		if err != nil {
 			log.Printf("Error while updating node coordinates: %s\n", err)
 		}
@@ -308,6 +378,13 @@ func GetNearestNeighbors() []NodeRegistration {
 	return nearestNeighbors
 }
 
+func GetRemoteOffloadingTarget() *NodeRegistration {
+	if remoteOffloadingTarget.Key != "" {
+		return &remoteOffloadingTarget
+	}
+	return nil
+}
+
 func GetFullNeighborInfo() map[string]*StatusInformation {
 	mutex.RLock()
 	defer mutex.RUnlock()

internal/registration/types.go

@@ -12,12 +12,12 @@ var IdRegistrationErr = errors.New("etcd error: could not complete the registrat
 type NodeRegistration struct {
 	node.NodeID
 	IPAddress      string
-	RemoteURL      string
+	APIPort        int
+	UDPPort        int
 	IsLoadBalancer bool
 }
 
 type StatusInformation struct {
-	Url                     string
 	AvailableWarmContainers map[string]int // <k, v> = <function name, warm container number>
 	AvailableMemMB          int64
 	AvailableCPUs           float64

internal/registration/udp.go

@@ -69,7 +69,6 @@ func handleUDPConnection(conn *net.UDPConn) {
 // TODO: this function should reuse the code in api.go for the /status API
 func getCurrentStatusInformation() (status []byte, err error) {
 	response := StatusInformation{
-		Url:                     SelfRegistration.RemoteURL,
 		AvailableWarmContainers: node.WarmStatus(),
 		AvailableMemMB:          node.Resources.AvailableMemMB,
 		AvailableCPUs:           node.Resources.AvailableCPUs,
@@ -80,8 +79,10 @@ func getCurrentStatusInformation() (status []byte, err error) {
 
 }
 
-func statusInfoRequest(hostname string) (info *StatusInformation, duration time.Duration) {
-	port := config.GetInt(config.LISTEN_UDP_PORT, 9876)
+func statusInfoRequest(peer *NodeRegistration) (info *StatusInformation, duration time.Duration) {
+
+	hostname := peer.IPAddress
+	port := peer.UDPPort
 	address := fmt.Sprintf("%s:%d", hostname, port)
 
 	remoteAddr, err := net.ResolveUDPAddr("udp", address)

internal/scheduling/offloading.go

@@ -27,7 +27,7 @@ func pickEdgeNodeForOffloading(r *scheduledRequest) (url string) {
 	}
 
 	randomItem := nearestNeighbors[rand.Intn(len(nearestNeighbors))]
-	return randomItem.RemoteURL
+	return randomItem.APIUrl()
 }
 
 func Offload(r *function.Request, serverUrl string) (function.ExecutionReport, error) {