Retreive Edge and remote execution times in ILPOffloadingPolicy

Author: grussorusso

internal/metrics/metrics.go

@@ -1,6 +1,7 @@
 package metrics
 
 import (
+	"fmt"
 	"log"
 
 	"net/http"
@@ -46,11 +47,28 @@ var (
 )
 
 type RetrievedMetrics struct {
-	Completions              map[string]float64
-	AvgExecutionTime         map[string]float64
-	AvgExecutionTimeAllNodes map[string]map[string]float64
-	AvgOutputSize            map[string]float64
-	BranchFrequency          map[string]map[string]float64
+	Completions map[string]float64
+	//AvgExecutionTime       map[string]float64
+	AvgRemoteExecutionTime map[string]float64
+	AvgEdgeExecutionTime   map[string]map[string]float64
+	AvgOutputSize          map[string]float64
+	BranchFrequency        map[string]map[string]float64
+}
+
+func (r RetrievedMetrics) String() string {
+	s := ""
+	s += "COMPLETIONS:\n"
+	s += fmt.Sprintf("  %v\n\n", r.Completions)
+	s += "REMOTE EXEC TIMES:\n"
+	s += fmt.Sprintf("  %v\n\n", r.AvgRemoteExecutionTime)
+	s += "EDGE EXEC TIMES:\n"
+	s += fmt.Sprintf("  %v\n\n", r.AvgEdgeExecutionTime)
+	s += "OUTPUT SIZE:\n"
+	s += fmt.Sprintf("  %v\n\n", r.AvgOutputSize)
+	s += "BRANCH FREQ:\n"
+	s += fmt.Sprintf("  %v\n\n", r.BranchFrequency)
+
+	return s
 }
 
 func Init() {

internal/metrics/retriever.go

@@ -3,6 +3,7 @@ package metrics
 import (
 	"context"
 	"fmt"
+	"github.com/serverledge-faas/serverledge/internal/registration"
 	"log"
 	"time"
 
@@ -176,21 +177,13 @@ func MetricsRetriever() {
 			}
 			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", err)
-			}
-			retrievedMetrics.AvgExecutionTime = avgFunDuration
-
-			// TODO: is this needed?
-			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", err)
-			}
-			retrievedMetrics.AvgExecutionTimeAllNodes = avgFunDurationAllNodes
+			//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", err)
+			//}
+			//retrievedMetrics.AvgExecutionTime = avgFunDuration
 
 			query = fmt.Sprintf("%s_sum{}/%s_count{}", OUTPUT_SIZE, OUTPUT_SIZE)
 			avgOutputSize, err := retrieveByFunction(query, api, ctx)
@@ -206,18 +199,28 @@ func MetricsRetriever() {
 			}
 			retrievedMetrics.BranchFrequency = frequencyPerTaskAndNextOne
 
+			// Execution time on Edge peers
+			localArea := registration.SelfRegistration.Area
+			query = fmt.Sprintf("%s_sum{node=~\"\\\\(%s\\\\).*\"}/%s_count{node=~\"\\\\(%s\\\\).*\"}",
+				EXECUTION_TIME, localArea, EXECUTION_TIME, localArea)
+			avgFunDurationAllNodes, err := retrieveByFunctionAndNode(query, api, ctx)
+			if err != nil {
+				log.Printf("Error in retrieveByFunction: %v", err)
+			}
+			retrievedMetrics.AvgEdgeExecutionTime = avgFunDurationAllNodes
+
 			// CLOUD
 			cloudArea := config.GetString(config.REGISTRY_REMOTE_AREA, "")
 			if cloudArea != "" {
-				log.Printf("Retrieved cloudarea %s", cloudArea)
 				query = fmt.Sprintf("%s_sum{node=~\"\\\\(%s\\\\).*\"}/%s_count{node=~\"\\\\(%s\\\\).*\"}",
 					EXECUTION_TIME, cloudArea, EXECUTION_TIME, cloudArea)
 				avgFunDuration, err := retrieveByFunction(query, api, ctx)
 				if err != nil {
 					log.Printf("Error in retrieveByFunction: %v", err)
 				}
-				fmt.Println(avgFunDuration)
-				// TODO: use this information
+				retrievedMetrics.AvgRemoteExecutionTime = avgFunDuration
+			} else {
+				retrievedMetrics.AvgRemoteExecutionTime = make(map[string]float64)
 			}
 
 			fmt.Println("All queries completed")

internal/workflow/ilp_offloading_policy.go

@@ -190,25 +190,32 @@ func (policy *IlpOffloadingPolicy) Evaluate(r *Request, p *Progress) (Offloading
 	}
 
 	// Execution Times
-	retrievedTimes := metrics.GetMetrics().AvgExecutionTimeAllNodes
+	retrievedMetrics := metrics.GetMetrics()
 	for tid, task := range r.W.Tasks {
 		ft, ok := task.(*FunctionTask)
 		if ok {
 			f, _ := function.GetFunction(ft.Func)
 			for _, n := range params.EdgeNodes {
-				nodeTimes, found := retrievedTimes[n]
+				nodeTimes, found := retrievedMetrics.AvgEdgeExecutionTime[n]
 				if !found {
-					params.ExecTime[tupleKey(string(tid), n)] = 1 // TODO: just guessing
+					params.ExecTime[tupleKey(string(tid), n)] = 1 // no data: just guessing
 					continue
 				}
 				t, found := nodeTimes[f.Name]
 				if found {
 					params.ExecTime[tupleKey(string(tid), n)] = t
 				} else {
-					params.ExecTime[tupleKey(string(tid), n)] = 1 // TODO: just guessing
+					params.ExecTime[tupleKey(string(tid), n)] = 1 // no data: just guessing
 				}
 			}
-			params.ExecTime[tupleKey(string(tid), CLOUD)] = 0.1 // TODO: how to retrieve cloud metrics? cloud nodes should use unique identifier??
+
+			// Cloud
+			t, found := retrievedMetrics.AvgRemoteExecutionTime[f.Name]
+			if found {
+				params.ExecTime[tupleKey(string(tid), CLOUD)] = t
+			} else {
+				params.ExecTime[tupleKey(string(tid), CLOUD)] = 1 // no data: just guessing
+			}
 		} else {
 			for _, n := range params.EdgeNodes {
 				params.ExecTime[tupleKey(string(tid), n)] = 0.0001

internal/workflow/workflow.go

@@ -12,8 +12,8 @@ import (
 
 	"log"
 
-	"github.com/serverledge-faas/serverledge/internal/config"
 	"github.com/serverledge-faas/serverledge/internal/client"
+	"github.com/serverledge-faas/serverledge/internal/config"
 	"github.com/serverledge-faas/serverledge/internal/metrics"
 	"golang.org/x/exp/slices"