Registry: Add CloudLatency

Author: matteonardelli

cmd/serverledge/main.go

@@ -93,6 +93,7 @@ func main() {
 		if err != nil {
 			log.Fatal(err)
 		}
+		registration.InitCloudMonitoring()
 	}
 
 	api.StartAPIServer(e)

internal/config/keys.go

@@ -50,6 +50,9 @@ const REG_NEARBY_INTERVAL = "registry.nearby.interval"
 // long period for general monitoring inside the area
 const REG_MONITORING_INTERVAL = "registry.monitoring.interval"
 
+// time interval for monitoring cloud latency
+const REG_CLOUD_MONITORING_INTERVAL = "registry.monitoring.cloud.interval"
+
 // registration TTL in seconds
 const REGISTRATION_TTL = "registry.ttl"
 

internal/registration/cloudMonitoring.go

@@ -0,0 +1,66 @@
+package registration
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"time"
+
+	"github.com/serverledge-faas/serverledge/internal/config"
+)
+
+func InitCloudMonitoring() {
+	Reg.CloudLatency = make(map[string]float64)
+	//complete monitoring phase at startup
+	monitorCloudLatency()
+	go runCloudMonitor()
+}
+
+func runCloudMonitor() {
+	cloudMonitorTicker := time.NewTicker(time.Duration(config.GetInt(config.REG_CLOUD_MONITORING_INTERVAL, 60)) * time.Second)
+	for {
+		select {
+		case <-cloudMonitorTicker.C:
+			monitorCloudLatency()
+		}
+	}
+}
+
+func tcpLatency(hostAndPort string) (float64, error) {
+	start := time.Now()
+	conn, err := net.DialTimeout("tcp", hostAndPort, 3*time.Second)
+	if err != nil {
+		return 0, err
+	}
+	_ = conn.Close()
+	return float64(time.Since(start).Milliseconds()), nil
+}
+
+func monitorCloudLatency() {
+	region := config.GetString(config.REGISTRY_AREA, "ROME")
+	cloudNodes, err := GetCloudNodes(region)
+	if err != nil {
+		log.Println(fmt.Printf("Error while retrieving cloud nodes: %v", err))
+		return
+	}
+
+	if len(cloudNodes) == 0 {
+		cloudAddress := config.GetString(config.CLOUD_URL, "")
+		if cloudAddress != "" {
+			cloudNodes["from-config"] = cloudAddress
+		}
+	}
+
+	for _, url := range cloudNodes {
+		// TODO: retrieve from offloadingLatency?
+		latency, latencyError := tcpLatency(url)
+		if latencyError != nil {
+			log.Printf("Unable to compute latency towards %v: %v\n", url, latencyError)
+			delete(Reg.CloudLatency, url)
+			continue
+		}
+		Reg.CloudLatency[url] = latency
+	}
+
+	log.Printf("Map of cloud node latency at the end of monitoring: %v\n", Reg.CloudLatency)
+}

internal/registration/types.go

@@ -18,6 +18,7 @@ type Registry struct {
 	RwMtx            trylock.Mutex
 	NearbyServersMap map[string]*StatusInformation
 	serversMap       map[string]*StatusInformation
+	CloudLatency     map[string]float64
 	etcdCh           chan bool
 }
 

internal/test/main_test.go

@@ -78,6 +78,7 @@ func testStartServerledge(isInCloud bool, outboundIp string) (*registration.Regi
 		if err != nil {
 			log.Fatal(err)
 		}
+		registration.InitCloudMonitoring()
 	}
 	// needed: if you call a function composition, internally will invoke each function
 	go api.StartAPIServer(e)