Improved cold start probability estimation

Author: grussorusso

internal/metrics/metrics.go

@@ -61,6 +61,7 @@ var (
 )
 
 type RetrievedMetrics struct {
+	EdgeColdStartProbability   map[string]map[string]float64
 	RemoteColdStartProbability map[string]float64
 	AvgRemoteExecutionTime     map[string]float64
 	AvgEdgeExecutionTime       map[string]map[string]float64
@@ -72,6 +73,8 @@ type RetrievedMetrics struct {
 
 func (r RetrievedMetrics) String() string {
 	s := ""
+	s += "EDGE COLD START PROB:\n"
+	s += fmt.Sprintf("  %v\n\n", r.EdgeColdStartProbability)
 	s += "REMOTE COLD START PROB:\n"
 	s += fmt.Sprintf("  %v\n\n", r.RemoteColdStartProbability)
 	s += "REMOTE EXEC TIMES:\n"

internal/metrics/retriever.go

@@ -203,6 +203,14 @@ func MetricsRetriever() {
 			}
 			retrievedMetrics.AvgEdgeInitTime = avgInitTimeAllNodes
 
+			query = fmt.Sprintf("%s{node=~\"\\\\(%s\\\\).*\"}/%s{node=~\"\\\\(%s\\\\).*\"}",
+				COLD_STARTS, localArea, COMPLETIONS, localArea)
+			coldStartProbPerFunction, err := retrieveByFunctionAndNode(query, api, ctx)
+			if err != nil {
+				log.Printf("Error in retrieveByFunction: %v", err)
+			}
+			retrievedMetrics.EdgeColdStartProbability = coldStartProbPerFunction
+
 			// CLOUD
 			cloudArea := config.GetString(config.REGISTRY_REMOTE_AREA, "")
 			if cloudArea != "" {

internal/workflow/remote_offloading_policy.go

@@ -285,8 +285,6 @@ func prepareParameters(r *Request, p *Progress) *remotePolicyParams {
 		params.DSBandwidth[CLOUD] = config.GetFloat(config.WORKFLOW_OFFLOADING_POLICY_CLOUD_TO_DATA_STORE_BANDWIDTH, dsBandwidth*10)
 	}
 
-	localWarmStatus := node.WarmStatus()
-
 	// Execution Times
 	retrievedMetrics := metrics.GetMetrics()
 	for tid, task := range r.W.Tasks {
@@ -312,38 +310,32 @@ func prepareParameters(r *Request, p *Progress) *remotePolicyParams {
 						params.ExecTime[tupleKey(string(tid), n)] = 0.01 // no data: just guessing
 					}
 				}
+				// Init Time
+				coldStartProb := 1.0
+				nodeProbs, found := retrievedMetrics.EdgeColdStartProbability[nId.String()]
+				if !found {
+					coldStartProb = 1.0
+				} else {
+					coldStartProb, found = nodeProbs[f.Name]
+					if !found || math.IsNaN(coldStartProb) || math.IsInf(coldStartProb, 1) {
+						coldStartProb = 1.0
+					}
+				}
 
 				// Init Times
 				initTimes, found := retrievedMetrics.AvgEdgeInitTime[nId.String()]
 				if !found {
 					// Unknown node
-					params.InitTime[tupleKey(string(tid), n)] = 0.01 // no data: just guessing
+					params.InitTime[tupleKey(string(tid), n)] = 0.01 * coldStartProb // no data: just guessing
 					continue
 				}
-
-				coldStart := false
-				if n == LOCAL {
-					warmCount, ok := localWarmStatus[f.Name]
-					if !ok || warmCount < 1 {
-						coldStart = true
-					}
+				t, found := initTimes[f.Name]
+				if found {
+					params.InitTime[tupleKey(string(tid), n)] = t * coldStartProb
 				} else {
-					warmCount, ok := nearbyServers[n].AvailableWarmContainers[f.Name]
-					if !ok || warmCount < 1 {
-						coldStart = true
-					}
+					params.InitTime[tupleKey(string(tid), n)] = 0.01 * coldStartProb // no data about init time: just guessing
 				}
 
-				if !coldStart {
-					params.InitTime[tupleKey(string(tid), n)] = 0
-				} else {
-					t, found := initTimes[f.Name]
-					if found {
-						params.InitTime[tupleKey(string(tid), n)] = t
-					} else {
-						params.InitTime[tupleKey(string(tid), n)] = 0.01 // no data: just guessing
-					}
-				}
 			}
 
 			if len(params.CloudNodes) > 0 {
@@ -358,9 +350,9 @@ func prepareParameters(r *Request, p *Progress) *remotePolicyParams {
 				// Init Time
 				coldStartProb := retrievedMetrics.RemoteColdStartProbability[f.Name]
 				if math.IsNaN(coldStartProb) || math.IsInf(coldStartProb, 1) {
+					log.Printf("Cloud cold start probability is invalid for %s, setting 1", f.Name)
 					coldStartProb = 1.0
 				} else if coldStartProb < 0.0 {
-					log.Printf("Cold start probability is negative: %f", coldStartProb)
 					coldStartProb = 0.0
 				}
 				t, found = retrievedMetrics.AvgRemoteInitTime[f.Name]
@@ -476,6 +468,5 @@ func computeDecisionFromPlacement(placement taskPlacement, p *Progress, r *Reque
 	}
 
 	decision := OffloadingDecision{true, remoteNodeReg.APIUrl(), plan}
-	log.Printf("Decision: %v\n", decision)
 	return decision
 }