Load Balancer registration

Author: grussorusso

cmd/lb/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"github.com/serverledge-faas/serverledge/internal/node"
 	"log"
 	"os"
 	"os/signal"
@@ -14,7 +15,6 @@ import (
 	"github.com/serverledge-faas/serverledge/internal/config"
 	"github.com/serverledge-faas/serverledge/internal/lb"
 	"github.com/serverledge-faas/serverledge/internal/registration"
-	"github.com/serverledge-faas/serverledge/utils"
 )
 
 func registerTerminationHandler(e *echo.Echo) {
@@ -44,19 +44,12 @@ func main() {
 	}
 	config.ReadConfiguration(configFileName)
 
-	// TODO: split Area in Region + Type (e.g., cloud/lb/edge)
-	region := config.GetString(config.REGISTRY_AREA, "ROME")
-	registry := &registration.Registry{Area: "lb/" + region}
+	myArea := config.GetString(config.REGISTRY_AREA, "ROME")
+	node.LocalNode = node.NewIdentifier(myArea)
 
-	defaultAddressStr := "127.0.0.1"
-	address, err := utils.GetOutboundIp()
-	if err == nil {
-		defaultAddressStr = address.String()
-	}
-	registration.RegisteredLocalIP = config.GetString(config.API_IP, defaultAddressStr)
-
-	if _, err := registry.RegisterToEtcd(); err != nil {
-		log.Printf("Could not register to Etcd: %v\n", err)
+	err := registration.RegisterLoadBalancer()
+	if err != nil {
+		log.Fatal(err)
 	}
 
 	e := echo.New()
@@ -66,5 +59,5 @@ func main() {
 	// Register a signal handler to cleanup things on termination
 	registerTerminationHandler(e)
 
-	lb.StartReverseProxy(e, region)
+	lb.StartReverseProxy(e, myArea)
 }

cmd/serverledge/main.go

@@ -33,7 +33,7 @@ func main() {
 	myArea := config.GetString(config.REGISTRY_AREA, "ROME")
 	node.LocalNode = node.NewIdentifier(myArea)
 
-	err := registration.RegisterToEtcd()
+	err := registration.RegisterNode()
 	if err != nil {
 		log.Fatal(err)
 	}

internal/api/api.go

@@ -238,6 +238,7 @@ func GetServerStatus(c echo.Context) error {
 		loadAvgValues = []float64{loadAvg.LoadAverage1, loadAvg.LoadAverage5, loadAvg.LoadAverage10}
 	}
 
+	// TODO: use a different type
 	response := registration.StatusInformation{
 		Url:                     registration.SelfRegistration.RemoteURL,
 		AvailableWarmContainers: node.WarmStatus(),

internal/lb/lb.go

@@ -9,10 +9,10 @@ import (
 	"os"
 	"time"
 
-	"github.com/serverledge-faas/serverledge/internal/config"
-	"github.com/serverledge-faas/serverledge/internal/registration"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
+	"github.com/serverledge-faas/serverledge/internal/config"
+	"github.com/serverledge-faas/serverledge/internal/registration"
 )
 
 var currentTargets []*middleware.ProxyTarget
@@ -42,20 +42,20 @@ func StartReverseProxy(e *echo.Echo, region string) {
 }
 
 func getTargets(region string) ([]*middleware.ProxyTarget, error) {
-	cloudNodes, err := registration.GetCloudNodes(region)
+	cloudNodes, err := registration.GetAllInArea(region, false)
 	if err != nil {
 		return nil, err
 	}
 
 	targets := make([]*middleware.ProxyTarget, 0, len(cloudNodes))
-	for _, addr := range cloudNodes {
-		log.Printf("Found target: %v\n", addr)
+	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(addr)
+		parsedUrl, err := url.Parse(target.RemoteURL)
 		if err != nil {
 			return nil, err
 		}
-		targets = append(targets, &middleware.ProxyTarget{Name: addr, URL: parsedUrl})
+		targets = append(targets, &middleware.ProxyTarget{Name: target.Key, URL: parsedUrl})
 	}
 
 	log.Printf("Found %d targets\n", len(targets))

internal/metrics/retriever.go

@@ -121,7 +121,7 @@ func MetricsRetriever() {
 	api := v1.NewAPI(client)
 	ctx := context.Background()
 
-	ticker := time.NewTicker(time.Duration(config.GetInt(config.METRICS_RETRIEVER_INTERVAL, 5)) * time.Second)
+	ticker := time.NewTicker(time.Duration(config.GetInt(config.METRICS_RETRIEVER_INTERVAL, 60)) * time.Second)
 	defer ticker.Stop()
 
 	for {
@@ -131,34 +131,34 @@ func MetricsRetriever() {
 			query := fmt.Sprintf("%s{node=\"%s\"}", COMPLETIONS, node.LocalNode)
 			completionsPerFunction, err := retrieveByFunction(query, api, ctx)
 			if err != nil {
-				log.Printf("Error in retrieveByFunction: %v\n", err)
+				log.Printf("Error in retrieveByFunction: %v", err)
 			}
 			retrievedMetrics.Completions = completionsPerFunction
 
 			query = fmt.Sprintf("%s_sum{node=\"%s\"}/%s_count{node=\"%s\"}",
 				EXECUTION_TIME, node.LocalNode, EXECUTION_TIME, node.LocalNode)
 			avgFunDuration, err := retrieveByFunction(query, api, ctx)
 			if err != nil {
-				log.Printf("Error in retrieveByFunction: %v\n", err)
+				log.Printf("Error in retrieveByFunction: %v", err)
 			}
 			retrievedMetrics.AvgExecutionTime = avgFunDuration
 
 			query = fmt.Sprintf("%s_sum{}/%s_count{}", EXECUTION_TIME, EXECUTION_TIME)
 			avgFunDurationAllNodes, err := retrieveByFunctionAndNode(query, api, ctx)
 			if err != nil {
-				log.Printf("Error in retrieveByFunction: %v\n", err)
+				log.Printf("Error in retrieveByFunction: %v", err)
 			}
 			retrievedMetrics.AvgExecutionTimeAllNodes = avgFunDurationAllNodes
 
 			query = fmt.Sprintf("%s_sum{}/%s_count{}", OUTPUT_SIZE, OUTPUT_SIZE)
 			avgOutputSize, err := retrieveByFunction(query, api, ctx)
 			if err != nil {
-				log.Printf("Error in retrieveByFunction: %v\n", err)
+				log.Printf("Error in retrieveByFunction: %v", err)
 			}
 			retrievedMetrics.AvgOutputSize = avgOutputSize
 
-			fmt.Println("All queries completed")
-			fmt.Println(retrievedMetrics)
+			//fmt.Println("All queries completed")
+			//fmt.Println(retrievedMetrics)
 
 		}
 	}