[WIP] Refactored registration package

Author: grussorusso

cmd/serverledge/main.go

@@ -9,16 +9,14 @@ import (
 	"os/signal"
 	"time"
 
+	"github.com/labstack/echo/v4"
 	"github.com/serverledge-faas/serverledge/internal/api"
 	"github.com/serverledge-faas/serverledge/internal/config"
 	"github.com/serverledge-faas/serverledge/internal/metrics"
 	"github.com/serverledge-faas/serverledge/internal/node"
 	"github.com/serverledge-faas/serverledge/internal/registration"
 	"github.com/serverledge-faas/serverledge/internal/scheduling"
 	"github.com/serverledge-faas/serverledge/internal/telemetry"
-	"github.com/serverledge-faas/serverledge/utils"
-
-	"github.com/labstack/echo/v4"
 )
 
 func main() {
@@ -32,32 +30,13 @@ func main() {
 	api.CacheSetup()
 
 	// register to etcd, this way server is visible to the others under a given local area
-	registry := new(registration.Registry)
-	isInCloud := config.GetBool(config.IS_IN_CLOUD, false)
-	if isInCloud {
-		registry.Area = "cloud/" + config.GetString(config.REGISTRY_AREA, "ROME")
-	} else {
-		registry.Area = config.GetString(config.REGISTRY_AREA, "ROME")
-	}
-	// before register checkout other servers into the local area
-	//todo use this info later on; future work with active remote server selection
-	_, err := registry.GetAll(true)
-	if err != nil {
-		log.Fatal(err)
-	}
+	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)
-
-	myKey, err := registry.RegisterToEtcd()
+	err := registration.RegisterToEtcd()
 	if err != nil {
 		log.Fatal(err)
 	}
-	node.NodeIdentifier = myKey
 
 	metrics.Init()
 
@@ -83,13 +62,14 @@ func main() {
 	e := echo.New()
 
 	// Register a signal handler to cleanup things on termination
-	api.RegisterTerminationHandler(registry, e)
+	api.RegisterTerminationHandler(e)
 
 	schedulingPolicy := api.CreateSchedulingPolicy()
 	go scheduling.Run(schedulingPolicy)
 
+	isInCloud := config.GetBool(config.IS_IN_CLOUD, false)
 	if !isInCloud {
-		err = registration.InitEdgeMonitoring(registry)
+		err = registration.StartMonitoring()
 		if err != nil {
 			log.Fatal(err)
 		}

internal/api/api.go

@@ -12,7 +12,6 @@ import (
 	"time"
 
 	"github.com/serverledge-faas/serverledge/internal/client"
-	"github.com/serverledge-faas/serverledge/internal/config"
 	"github.com/serverledge-faas/serverledge/internal/container"
 	"github.com/serverledge-faas/serverledge/internal/function"
 	"github.com/serverledge-faas/serverledge/internal/node"
@@ -74,7 +73,7 @@ func InvokeFunction(c echo.Context) error {
 	r.Async = invocationRequest.Async
 	r.ReturnOutput = invocationRequest.ReturnOutput
 
-	reqId := fmt.Sprintf("%s-%s%d", fun, node.NodeIdentifier[len(node.NodeIdentifier)-5:], r.Arrival.Nanosecond())
+	reqId := fmt.Sprintf("%s-%s%d", funcName, node.LocalNode.String()[len(node.LocalNode.String())-5:], r.Arrival.Nanosecond())
 	r.Ctx = context.WithValue(context.Background(), "ReqId", reqId)
 
 	// Tracing
@@ -233,22 +232,18 @@ func GetServerStatus(c echo.Context) error {
 	node.Resources.RLock()
 	defer node.Resources.RUnlock()
 
-	portNumber := config.GetInt("api.port", 1323)
-
-	url := fmt.Sprintf("http://%s:%d", registration.RegisteredLocalIP, portNumber)
-
 	loadAvg, err := loadavg.Parse()
 	loadAvgValues := []float64{-1.0, -1.0, -1.0}
 	if err == nil {
 		loadAvgValues = []float64{loadAvg.LoadAverage1, loadAvg.LoadAverage5, loadAvg.LoadAverage10}
 	}
 
 	response := registration.StatusInformation{
-		Url:                     url,
+		Url:                     registration.SelfRegistration.RemoteURL,
 		AvailableWarmContainers: node.WarmStatus(),
 		AvailableMemMB:          node.Resources.AvailableMemMB,
 		AvailableCPUs:           node.Resources.AvailableCPUs,
-		Coordinates:             *registration.Reg.Client.GetCoordinate(),
+		Coordinates:             *registration.VivaldiClient.GetCoordinate(),
 		LoadAvg:                 loadAvgValues,
 	}
 

internal/api/server.go

@@ -76,7 +76,7 @@ func CacheSetup() {
 	cache.GetCacheInstance()
 }
 
-func RegisterTerminationHandler(r *registration.Registry, e *echo.Echo) {
+func RegisterTerminationHandler(e *echo.Echo) {
 	c := make(chan os.Signal)
 	signal.Notify(c, os.Interrupt)
 
@@ -87,7 +87,7 @@ func RegisterTerminationHandler(r *registration.Registry, e *echo.Echo) {
 			node.ShutdownAllContainers()
 
 			// deregister from etcd; server should be unreachable
-			err := r.Deregister()
+			err := registration.Deregister()
 			if err != nil {
 				log.Fatal(err)
 			}

internal/api/workflow.go

@@ -213,7 +213,7 @@ func InvokeWorkflow(e echo.Context) error {
 	req.Async = clientReq.Async
 	req.Plan = nil
 	req.Resuming = false
-	req.Id = fmt.Sprintf("%v-%s%d", wflow.Name, node.NodeIdentifier[len(node.NodeIdentifier)-5:], req.Arrival.Nanosecond())
+	req.Id = fmt.Sprintf("%v-%s%d", wflow.Name, node.LocalNode.String()[len(node.LocalNode.String())-5:], req.Arrival.Nanosecond())
 	req.ExecReport.Reports = map[string]*function.ExecutionReport{}
 
 	return handleWorkflowInvocation(e, req)

internal/config/keys.go

@@ -44,15 +44,15 @@ const IS_IN_CLOUD = "cloud"
 // the area wich the server belongs to
 const REGISTRY_AREA = "registry.area"
 
+// the area that acts as "remote cloud" for this node
+const REGISTRY_REMOTE_AREA = "registry.remote.area"
+
 // short period: retrieve information about nearby edge-servers
 const REG_NEARBY_INTERVAL = "registry.nearby.interval"
 
 // long period for general monitoring inside the area
 const REG_MONITORING_INTERVAL = "registry.monitoring.interval"
 
-// registration TTL in seconds
-const REGISTRATION_TTL = "registry.ttl"
-
 // port for udp status listener
 const LISTEN_UDP_PORT = "registry.udp.port"
 

internal/metrics/metrics.go

@@ -67,10 +67,10 @@ func Init() {
 }
 
 func AddCompletedInvocation(funcName string) {
-	metricCompletions.With(prometheus.Labels{"function": funcName, "node": node.NodeIdentifier}).Inc()
+	metricCompletions.With(prometheus.Labels{"function": funcName, "node": node.LocalNode.String()}).Inc()
 }
 func AddFunctionDurationValue(funcName string, duration float64) {
-	metricExecutionTime.With(prometheus.Labels{"function": funcName, "node": node.NodeIdentifier}).Observe(duration)
+	metricExecutionTime.With(prometheus.Labels{"function": funcName, "node": node.LocalNode.String()}).Observe(duration)
 }
 func AddFunctionOutputSizeValue(funcName string, size float64) {
 	metricOutputSize.With(prometheus.Labels{"function": funcName}).Observe(size)

internal/metrics/retriever.go

@@ -128,15 +128,15 @@ func MetricsRetriever() {
 		select {
 		case <-ticker.C:
 
-			query := fmt.Sprintf("%s{node=\"%s\"}", COMPLETIONS, node.NodeIdentifier)
+			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)
 			}
 			retrievedMetrics.Completions = completionsPerFunction
 
 			query = fmt.Sprintf("%s_sum{node=\"%s\"}/%s_count{node=\"%s\"}",
-				EXECUTION_TIME, node.NodeIdentifier, EXECUTION_TIME, node.NodeIdentifier)
+				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)

internal/node/node.go

@@ -3,12 +3,29 @@ package node
 import (
 	"errors"
 	"fmt"
+	"github.com/lithammer/shortuuid"
+	"strconv"
 	"sync"
+	"time"
 )
 
 var OutOfResourcesErr = errors.New("not enough resources for function execution")
 
-var NodeIdentifier string
+type NodeID struct {
+	Area string
+	Key  string
+}
+
+var LocalNode NodeID
+
+func (n NodeID) String() string {
+	return fmt.Sprintf("%s/%s", n.Area, n.Key)
+}
+
+func NewIdentifier(area string) NodeID {
+	id := shortuuid.New() + strconv.FormatInt(time.Now().UnixNano(), 10)
+	return NodeID{Area: area, Key: id}
+}
 
 type NodeResources struct {
 	sync.RWMutex